wehub/paddlepaddle--paddle
1
0
Fork 0
Code Issues Pull Requests Actions 2 Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
paddlepaddle--paddle/test/legacy_test/test_reduce_op.py
T
wehub-resource-sync e25996e7db chore: import upstream snapshot with attribution
2026-07-13 12:40:42 +08:00

2828 lines
91 KiB
Python
Raw Permalink Blame History

# Copyright (c) 2018 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 unittest
import numpy as np
from op_test import (
OpTest,
convert_float_to_uint16,
get_device_place,
get_places,
is_custom_device,
skip_check_grad_ci,
)
from utils import dygraph_guard, static_guard
import paddle
from paddle import base
from paddle.base import core
from paddle.base.framework import in_pir_mode
class TestSumOp(OpTest):
def setUp(self):
self.init_dtype()
self.init_input()
self.init_attrs()
self.calc_output()
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.op_type = "reduce_sum"
self.prim_op_type = "prim"
self.inputs = {'X': self.x}
self.outputs = {'Out': self.out}
self.if_enable_cinn()
def init_dtype(self):
self.dtype = np.float64
def init_input(self):
self.x = np.random.random((5, 6, 10)).astype(self.dtype)
def init_attrs(self):
self.attrs = {'dim': [0]}
def if_enable_cinn(self):
pass
def calc_output(self):
self.out = self.x.sum(axis=tuple(self.attrs['dim']))
def test_check_output(self):
self.check_output(check_pir=True)
def test_check_grad(self):
self.check_grad(
['X'],
'Out',
check_prim=False,
check_pir=True,
check_prim_pir=True,
)
class TestComplexSumOP(TestSumOp):
def init_dtype(self):
self.dtype = np.complex128
def init_input(self):
self.x = np.random.random((3, 4)).astype(self.dtype)
def init_attrs(self):
self.attrs = {'dim': [0]}
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False)
class TestSumOp_ZeroDim(TestSumOp):
def init_attrs(self):
self.attrs = {'dim': []}
def init_input(self):
self.x = np.random.random([]).astype(self.dtype)
def calc_output(self):
self.out = self.x.sum(axis=None)
def test_check_grad(self):
self.check_grad(
['X'],
'Out',
check_pir=True,
check_prim=False,
check_prim_pir=True,
)
class TestSumOp5D(TestSumOp):
def init_input(self):
self.x = np.random.random((1, 2, 5, 6, 10)).astype(self.dtype)
def init_attrs(self):
self.attrs = {'dim': [0]}
class TestSumOp6D(TestSumOp):
def init_input(self):
self.x = np.random.random((1, 1, 2, 5, 6, 10)).astype(self.dtype)
def init_attrs(self):
self.attrs = {'dim': [0]}
class TestSumOp8D(TestSumOp):
def init_input(self):
self.x = np.random.random((1, 3, 1, 2, 1, 4, 3, 10)).astype(self.dtype)
def init_attrs(self):
self.attrs = {'dim': (0, 3)}
def test_check_output(self):
self.check_output(check_pir=True)
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_pir=True)
class TestSumOp_withInt(TestSumOp):
def init_input(self):
# ref to https://en.wikipedia.org/wiki/Half-precision_floating-point_format
# Precision limitations on integer values between 0 and 2048 can be exactly represented
self.x = np.random.randint(0, 30, (10, 10)).astype(self.dtype)
def init_attrs(self):
self.attrs = {'dim': (0, 1)}
def test_check_output(self):
self.check_output(check_pir=True)
def calc_gradient(self):
x = self.inputs["X"]
grad = np.ones(x.shape, dtype=x.dtype)
return (grad,)
def test_check_grad(self):
self.check_grad(
['X'],
'Out',
user_defined_grads=self.calc_gradient(),
check_prim=False,
check_prim_pir=True,
check_pir=True,
)
class TestSumOp3Dim(TestSumOp):
def init_input(self):
self.x = np.random.uniform(0, 0.1, (5, 6, 10)).astype(self.dtype)
def init_attrs(self):
self.attrs = {'dim': (0, 1, 2)}
def test_check_output(self):
self.check_output(check_pir=True)
def calc_gradient(self):
x = self.inputs["X"]
grad = np.ones(x.shape, dtype=x.dtype)
return (grad,)
def test_check_grad(self):
self.check_grad(
['X'],
'Out',
user_defined_grads=self.calc_gradient(),
check_prim=False,
check_prim_pir=True,
check_pir=True,
)
def create_test_fp16_class(parent):
@unittest.skipIf(
not (core.is_compiled_with_cuda() or is_custom_device()),
"core is not compiled with CUDA",
)
class TestSumOpFp16(parent):
def init_dtype(self):
self.dtype = np.float16
def test_check_output(self):
self.check_output(check_pir=True)
def test_check_grad(self):
self.check_grad(
['X'],
'Out',
check_prim=False,
check_prim_pir=True,
check_pir=True,
)
def create_test_fp16_class_cpu(parent):
class TestSumOpFp16CPU(parent):
def init_dtype(self):
self.dtype = np.float16
def test_check_output(self):
self.check_output(check_pir=True, rtol=1e-2, atol=1e-2)
def test_check_grad(self):
self.check_grad(
['X'],
'Out',
check_prim=False,
check_prim_pir=True,
check_pir=True,
)
class TestSumOp3D0size(TestSumOp3Dim):
def test_check_output(self):
self.check_output(check_pir=True, check_pir_onednn=True)
def calc_gradient(self):
x = self.inputs["X"]
grad = np.ones(x.shape, dtype=x.dtype)
return (grad,)
def test_check_grad(self):
self.check_grad(
['X'],
'Out',
user_defined_grads=self.calc_gradient(),
check_prim=False,
check_prim_pir=True,
check_pir=True,
check_pir_onednn=True,
)
class TestSumOp3D0size1(TestSumOp3D0size):
def init_input(self):
self.x = np.random.uniform(0, 0.1, (5, 0, 10)).astype(self.dtype)
def init_attrs(self):
self.attrs = {'dim': (0, 1, 2)}
class TestSumOp3D0size2(TestSumOp3D0size):
def init_input(self):
self.x = np.random.uniform(0, 0.1, (0, 6, 10)).astype(self.dtype)
def init_attrs(self):
self.attrs = {'dim': (0, 1, 2)}
class TestSumOp3D0size3(TestSumOp3D0size):
def init_input(self):
self.x = np.random.uniform(0, 0.1, (4, 6, 0)).astype(self.dtype)
def init_attrs(self):
self.attrs = {'dim': (0, 1, 2)}
create_test_fp16_class(TestSumOp)
create_test_fp16_class(TestSumOp_ZeroDim)
create_test_fp16_class(TestSumOp5D)
create_test_fp16_class(TestSumOp6D)
create_test_fp16_class(TestSumOp8D)
create_test_fp16_class(TestSumOp_withInt)
create_test_fp16_class(TestSumOp3Dim)
create_test_fp16_class_cpu(TestSumOp)
create_test_fp16_class_cpu(TestSumOp_ZeroDim)
create_test_fp16_class_cpu(TestSumOp5D)
create_test_fp16_class_cpu(TestSumOp6D)
create_test_fp16_class_cpu(TestSumOp8D)
create_test_fp16_class_cpu(TestSumOp_withInt)
create_test_fp16_class_cpu(TestSumOp3Dim)
def create_test_bf16_class(parent):
@unittest.skipIf(
not core.is_compiled_with_cuda() or paddle.is_compiled_with_rocm(),
"core is not compiled with CUDA",
)
class TestSumOpBf16(parent):
def setUp(self):
self.inputs = {'X': convert_float_to_uint16(self.x)}
self.outputs = {'Out': convert_float_to_uint16(self.out)}
self.enable_cinn = False
def init_dtype(self):
self.dtype = np.uint16
def test_check_output(self):
place = get_device_place()
self.check_output_with_place(place, check_pir=True)
def test_check_grad(self):
place = get_device_place()
self.check_grad_with_place(
place,
['X'],
'Out',
user_defined_grads=self.gradient,
check_prim=False,
check_prim_pir=True,
check_pir=True,
)
def calc_gradient(self):
x = self.x
grad = np.ones(x.shape, dtype=x.dtype)
return [grad]
create_test_bf16_class(TestSumOp)
create_test_bf16_class(TestSumOp_ZeroDim)
create_test_bf16_class(TestSumOp5D)
create_test_bf16_class(TestSumOp6D)
create_test_bf16_class(TestSumOp8D)
create_test_bf16_class(TestSumOp_withInt)
create_test_bf16_class(TestSumOp3Dim)
class TestSumAPIZeroDimKeepDim(unittest.TestCase):
def setUp(self):
np.random.seed(123)
paddle.enable_static()
self.places = get_places()
def test_static(self):
for place in self.places:
main = paddle.static.Program()
startup = paddle.static.Program()
with base.program_guard(main, startup):
input = paddle.static.data(
name="input", shape=[0, 0], dtype="float32"
)
result = paddle.sum(x=input, keepdim=True)
input_np = np.random.rand(0, 0).astype("float32")
exe = paddle.static.Executor(place)
fetches = exe.run(
main,
feed={"input": input_np},
fetch_list=[result],
)
self.assertEqual(fetches[0].shape, (1, 1))
np.allclose(fetches[0], np.sum(input_np, keepdims=True))
def test_dygraph(self):
paddle.disable_static()
for place in self.places:
with base.dygraph.guard(place):
np_x = np.random.rand(0, 0).astype("float32")
x = paddle.to_tensor(np_x)
out1 = paddle.sum(x, keepdim=True)
np_out1 = out1.numpy()
expect_res1 = np.sum(np_x, keepdims=True)
np.allclose(np_out1, expect_res1)
out2 = paddle.sum(x, axis=0, keepdim=True)
np_out2 = out2.numpy()
expect_res2 = np.sum(np_x, axis=0, keepdims=True)
np.allclose(np_out2, expect_res2)
out3 = paddle.sum(x, axis=-1, keepdim=True)
np_out3 = out3.numpy()
expect_res3 = np.sum(np_x, axis=-1, keepdims=True)
np.allclose(np_out3, expect_res3)
paddle.enable_static()
@skip_check_grad_ci(
reason="reduce_max is discontinuous non-derivable function,"
" its gradient check is not supported by unittest framework."
)
class TestMaxOp(OpTest):
"""Remove Max with subgradient from gradient check to confirm the success of CI."""
def setUp(self):
self.op_type = "reduce_max"
self.prim_op_type = "prim"
self.python_api = paddle.max
self.public_python_api = paddle.max
self.inputs = {'X': np.random.random((5, 6, 10)).astype("float64")}
self.attrs = {'dim': [-1]}
self.outputs = {
'Out': self.inputs['X'].max(axis=tuple(self.attrs['dim']))
}
def test_check_output(self):
self.check_output(check_pir=True)
def test_check_grad(self):
# only composite op support gradient check of reduce_max
self.check_grad(
['X'],
'Out',
check_prim=False,
only_check_prim=True,
check_pir=True,
)
class TestMaxOp_ZeroDim(OpTest):
"""Remove Max with subgradient from gradient check to confirm the success of CI."""
def setUp(self):
self.op_type = "reduce_max"
self.prim_op_type = "prim"
self.python_api = paddle.max
self.public_python_api = paddle.max
self.if_enable_cinn()
self.init_inputs_and_outputs()
def if_enable_cinn(self):
self.enable_cinn = False
def init_inputs_and_outputs(self):
self.inputs = {'X': np.random.random([]).astype("float64")}
self.attrs = {'dim': []}
self.outputs = {
'Out': self.inputs['X'].max(axis=tuple(self.attrs['dim']))
}
def test_check_output(self):
self.check_output(check_pir=True)
def test_check_grad(self):
# only composite op support gradient check of reduce_max
self.check_grad(
['X'],
'Out',
check_prim=False,
only_check_prim=True,
check_pir=True,
)
class TestMaxOp_ZeroDim1(TestMaxOp_ZeroDim):
def init_inputs_and_outputs(self):
self.inputs = {'X': np.random.random([5]).astype("float64")}
self.attrs = {'dim': [0]}
self.outputs = {'Out': self.inputs['X'].max(axis=(0,))}
class TestMaxOp_ZeroDim2(TestMaxOp_ZeroDim1):
def init_inputs_and_outputs(self):
self.inputs = {'X': np.random.random([5, 20]).astype("float64")}
self.attrs = {'dim': [0, 1]}
self.outputs = {'Out': self.inputs['X'].max(axis=(0, 1))}
class TestMaxFP32Op(OpTest):
"""Remove Max with subgradient from gradient check to confirm the success of CI."""
def setUp(self):
self.op_type = "reduce_max"
self.prim_op_type = "prim"
self.python_api = paddle.max
self.public_python_api = paddle.max
self.init_dtype()
self.if_enable_cinn()
if self.dtype == np.uint16:
x = np.random.random((5, 6, 10)).astype(np.float32)
self.inputs = {'X': convert_float_to_uint16(x)}
else:
x = np.random.random((5, 6, 10)).astype(self.dtype)
self.inputs = {'X': x}
self.attrs = {'dim': [-1], 'keep_dim': True}
out = x.max(axis=tuple(self.attrs['dim']), keepdims=True)
if self.dtype == np.uint16:
self.outputs = {'Out': convert_float_to_uint16(out)}
else:
self.outputs = {'Out': out}
def if_enable_cinn(self):
pass
def test_check_output(self):
self.check_output(check_pir=True)
def test_check_grad(self):
# only composite op support gradient check of reduce_max
self.check_grad(
['X'],
'Out',
check_prim=False,
only_check_prim=True,
check_pir=True,
)
def init_dtype(self):
self.dtype = np.float32
class TestMaxFP16Op(TestMaxFP32Op):
def init_dtype(self):
self.dtype = np.float16
@unittest.skipIf(
not core.is_compiled_with_cuda()
or paddle.is_compiled_with_rocm()
or not core.is_bfloat16_supported(get_device_place()),
"core is not compiled with CUDA or not support the bfloat16",
)
class TestMaxBF16Op(TestMaxFP32Op):
def init_dtype(self):
self.dtype = np.uint16
def if_enable_cinn(self):
self.enable_cinn = False
def test_check_output(self):
self.check_output_with_place(get_device_place(), check_pir=True)
def test_check_grad(self):
# only composite op support gradient check of reduce_max
self.check_grad_with_place(
get_device_place(),
['X'],
'Out',
check_prim=False,
only_check_prim=True,
check_pir=True,
)
@skip_check_grad_ci(
reason="reduce_min is discontinuous non-derivable function,"
" its gradient check is not supported by unittest framework."
)
class TestMinOp(OpTest):
"""Remove Min with subgradient from gradient check to confirm the success of CI."""
def setUp(self):
self.op_type = "reduce_min"
self.python_api = paddle.min
self.inputs = {'X': np.random.random((5, 6, 10)).astype("float64")}
self.attrs = {'dim': [2]}
self.outputs = {
'Out': self.inputs['X'].min(axis=tuple(self.attrs['dim']))
}
def test_check_output(self):
self.check_output(check_pir=True)
class TestMinOp_ZeroDim(OpTest):
"""Remove Min with subgradient from gradient check to confirm the success of CI."""
def setUp(self):
self.op_type = "reduce_min"
self.python_api = paddle.min
self.inputs = {'X': np.random.random([]).astype("float64")}
self.attrs = {'dim': []}
self.outputs = {
'Out': self.inputs['X'].min(axis=tuple(self.attrs['dim']))
}
def test_check_output(self):
self.check_output(check_pir=True)
class TestMin6DOp(OpTest):
"""Remove Min with subgradient from gradient check to confirm the success of CI."""
def setUp(self):
self.op_type = "reduce_min"
self.python_api = paddle.min
self.inputs = {
'X': np.random.random((2, 4, 3, 5, 6, 10)).astype("float64")
}
self.attrs = {'dim': [2, 4]}
self.outputs = {
'Out': self.inputs['X'].min(axis=tuple(self.attrs['dim']))
}
def test_check_output(self):
self.check_output(check_pir=True)
class TestMin8DOp(OpTest):
"""Remove Min with subgradient from gradient check to confirm the success of CI."""
def setUp(self):
self.op_type = "reduce_min"
self.python_api = paddle.min
self.inputs = {
'X': np.random.random((2, 4, 3, 5, 6, 3, 2, 4)).astype("float64")
}
self.attrs = {'dim': [2, 3, 4]}
self.outputs = {
'Out': self.inputs['X'].min(axis=tuple(self.attrs['dim']))
}
def test_check_output(self):
self.check_output(check_pir=True)
@skip_check_grad_ci(
reason="reduce_min is discontinuous non-derivable function,"
" its gradient check is not supported by unittest framework."
)
@unittest.skipIf(
paddle.is_compiled_with_rocm(), "ROCm doesn't have FP16 reduce_min kernel"
)
class TestMinFP16Op(OpTest):
"""Remove Min with subgradient from gradient check to confirm the success of CI."""
def setUp(self):
self.op_type = "reduce_min"
self.python_api = paddle.min
self.public_python_api = paddle.min
self.init_dtype()
if self.dtype == np.uint16:
x = np.random.random((5, 6, 10)).astype(np.float32)
self.inputs = {'X': convert_float_to_uint16(x)}
else:
x = np.random.random((5, 6, 10)).astype(self.dtype)
self.inputs = {'X': x}
self.attrs = {'dim': [2], 'keep_dim': True}
out = x.min(axis=tuple(self.attrs['dim']), keepdims=True)
if self.dtype == np.uint16:
self.outputs = {'Out': convert_float_to_uint16(out)}
else:
self.outputs = {'Out': out}
def init_dtype(self):
self.dtype = np.float16
def test_check_output(self):
self.check_output(check_pir=True)
@unittest.skipIf(
not core.is_compiled_with_cuda()
or paddle.is_compiled_with_rocm()
or not core.is_bfloat16_supported(get_device_place()),
"core is not compiled with CUDA or not support the bfloat16",
)
class TestMinBF16Op(TestMinFP16Op):
def init_dtype(self):
self.dtype = np.uint16
def test_check_output(self):
self.check_output_with_place(get_device_place(), check_pir=True)
def raw_reduce_prod(x, dim=[0], keep_dim=False):
return paddle.prod(x, dim, keep_dim)
class TestProdOp(OpTest):
def setUp(self):
self.op_type = "reduce_prod"
self.python_api = raw_reduce_prod
self.public_python_api = raw_reduce_prod
self.prim_op_type = "prim"
self.init_data_type()
self.init_inputs_and_outputs()
self.if_enable_cinn()
def init_inputs_and_outputs(self):
self.inputs = {'X': np.random.random((5, 6, 10)).astype(self.data_type)}
self.outputs = {'Out': self.inputs['X'].prod(axis=0)}
def init_data_type(self):
self.data_type = (
"float32" if core.is_compiled_with_rocm() else "float64"
)
def if_enable_cinn(self):
pass
def test_check_output(self):
self.check_output(check_pir=True)
def test_check_grad(self):
self.check_grad(
['X'], 'Out', check_prim=False, check_pir=True, check_prim_pir=True
)
@unittest.skipIf(
not (paddle.is_compiled_with_cuda() or is_custom_device()),
"FP16 test runs only on GPU",
)
class TestProdFP16OP(TestProdOp):
def init_data_type(self):
self.data_type = "float16"
def test_check_output(self):
self.check_output_with_place(place=get_device_place(), check_pir=True)
def test_check_grad(self):
self.check_grad_with_place(
get_device_place(),
['X'],
'Out',
check_prim=False,
check_pir=True,
check_prim_pir=True,
)
@unittest.skipIf(
not core.is_compiled_with_cuda()
or paddle.is_compiled_with_rocm()
or not core.is_bfloat16_supported(get_device_place()),
"core is not compiled with CUDA or not support the bfloat16",
)
class TestProdBFP16OP(TestProdOp):
def init_data_type(self):
self.data_type = np.uint16
def init_inputs_and_outputs(self):
x = np.random.random((5, 6, 10)).astype("float32")
out = x.prod(axis=0)
self.inputs = {'X': convert_float_to_uint16(x)}
self.outputs = {'Out': convert_float_to_uint16(out)}
def if_enable_cinn(self):
self.enable_cinn = False
def test_check_output(self):
self.check_output_with_place(place=get_device_place(), check_pir=True)
def test_check_grad(self):
self.check_grad_with_place(
get_device_place(),
['X'],
'Out',
check_prim=False,
check_pir=True,
check_prim_pir=True,
)
class TestProdOpFp64(TestProdOp):
def init_data_type(self):
self.data_type = "float64"
class TestProdOp_ZeroDim(OpTest):
def setUp(self):
self.python_api = raw_reduce_prod
self.public_python_api = raw_reduce_prod
self.op_type = "reduce_prod"
self.prim_op_type = "prim"
self.init_inputs_and_outputs()
# 0-D tensor doesn't support in cinn
self.enable_cinn = False
def init_inputs_and_outputs(self):
self.inputs = {'X': np.random.random([]).astype("float64")}
self.outputs = {'Out': self.inputs['X'].prod()}
self.attrs = {'dim': [], 'reduce_all': True}
def test_check_output(self):
self.check_output(check_pir=True)
def test_check_grad(self):
self.check_grad(
['X'], 'Out', check_prim=False, check_pir=True, check_prim_pir=True
)
class TestProdOp_ZeroDim1(TestProdOp):
def setUp(self):
self.python_api = paddle.prod
self.public_python_api = paddle.prod
self.op_type = "reduce_prod"
self.prim_op_type = "prim"
self.init_inputs_and_outputs()
# 0-D tensor doesn't support in cinn
self.enable_cinn = False
def init_inputs_and_outputs(self):
self.inputs = {'X': np.random.random([100]).astype("float64")}
self.outputs = {'Out': self.inputs['X'].prod()}
self.attrs = {'dim': [], 'reduce_all': True}
class TestProdOp_ZeroDim2(TestProdOp_ZeroDim1):
def init_inputs_and_outputs(self):
self.inputs = {'X': np.random.random([5, 6, 10]).astype("float64")}
self.outputs = {'Out': self.inputs['X'].prod()}
self.attrs = {'dim': [], 'reduce_all': True}
class TestProd6DOp(OpTest):
def setUp(self):
self.op_type = "reduce_prod"
self.python_api = raw_reduce_prod
self.public_python_api = raw_reduce_prod
self.prim_op_type = "prim"
self.init_data_type()
self.init_inputs_and_outputs()
self.if_enable_cinn()
def init_data_type(self):
self.data_type = (
"float32" if core.is_compiled_with_rocm() else "float64"
)
def init_inputs_and_outputs(self):
self.inputs = {
'X': np.random.random((5, 6, 2, 3, 4, 2)).astype(self.data_type)
}
self.attrs = {'dim': [2, 3, 4]}
self.outputs = {
'Out': self.inputs['X'].prod(axis=tuple(self.attrs['dim']))
}
def if_enable_cinn(self):
pass
def test_check_output(self):
self.check_output(check_pir=True)
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False, check_pir=True)
@unittest.skipIf(
not (paddle.is_compiled_with_cuda() or is_custom_device()),
"FP16 test runs only on GPU",
)
class TestProd6DFP16OP(TestProd6DOp):
def init_data_type(self):
self.data_type = "float16"
def test_check_output(self):
self.check_output_with_place(place=get_device_place(), check_pir=True)
def test_check_grad(self):
self.check_grad_with_place(
get_device_place(), ['X'], 'Out', check_prim=False, check_pir=True
)
@unittest.skipIf(
not core.is_compiled_with_cuda()
or paddle.is_compiled_with_rocm()
or not core.is_bfloat16_supported(get_device_place()),
"core is not compiled with CUDA or not support the bfloat16",
)
class TestProd6DBFP16OP(TestProd6DOp):
def init_data_type(self):
self.data_type = np.uint16
def init_inputs_and_outputs(self):
x = np.random.random((5, 6, 2, 3, 4, 2)).astype("float32")
self.attrs = {'dim': [2, 3, 4]}
out = x.prod(axis=tuple(self.attrs['dim']))
self.inputs = {'X': convert_float_to_uint16(x)}
self.outputs = {'Out': convert_float_to_uint16(out)}
def if_enable_cinn(self):
self.enable_cinn = False
def test_check_output(self):
self.check_output_with_place(place=get_device_place(), check_pir=True)
def test_check_grad(self):
self.check_grad_with_place(
get_device_place(), ['X'], 'Out', check_prim=False, check_pir=True
)
class TestProd8DOp(OpTest):
def setUp(self):
self.op_type = "reduce_prod"
self.python_api = raw_reduce_prod
self.public_python_api = raw_reduce_prod
self.init_data_type()
self.init_inputs_and_outputs()
def init_inputs_and_outputs(self):
self.inputs = {
'X': np.random.random((2, 5, 3, 2, 2, 3, 4, 2)).astype(
self.data_type
)
}
self.attrs = {'dim': [2, 3, 4]}
self.outputs = {
'Out': self.inputs['X'].prod(axis=tuple(self.attrs['dim']))
}
def init_data_type(self):
self.data_type = (
"float32" if core.is_compiled_with_rocm() else "float64"
)
def test_check_output(self):
self.check_output(check_pir=True)
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_pir=True)
@unittest.skipIf(
not (paddle.is_compiled_with_cuda() or is_custom_device()),
"FP16 test runs only on GPU",
)
class TestProd8DFP16OP(TestProd8DOp):
def init_data_type(self):
self.data_type = "float16"
def test_check_output(self):
self.check_output_with_place(place=get_device_place(), check_pir=True)
def test_check_grad(self):
self.check_grad_with_place(
get_device_place(), ['X'], 'Out', check_pir=True
)
@unittest.skipIf(
not core.is_compiled_with_cuda()
or paddle.is_compiled_with_rocm()
or not core.is_bfloat16_supported(get_device_place()),
"core is not compiled with CUDA or not support the bfloat16",
)
class TestProd8DBFP16OP(TestProd8DOp):
def init_data_type(self):
self.data_type = np.uint16
def init_inputs_and_outputs(self):
x = np.random.random((2, 5, 3, 2, 2, 3, 4, 2)).astype("float32")
self.attrs = {'dim': [2, 3, 4]}
out = x.prod(axis=tuple(self.attrs['dim']))
self.inputs = {'X': convert_float_to_uint16(x)}
self.outputs = {'Out': convert_float_to_uint16(out)}
def test_check_output(self):
self.check_output_with_place(place=get_device_place(), check_pir=True)
def test_check_grad(self):
self.check_grad_with_place(
get_device_place(), ['X'], 'Out', check_pir=True
)
def reduce_all_wrapper(x, axis=None, keepdim=False, reduce_all=True, name=None):
return paddle.all(x, axis, keepdim, name)
class TestAllOp(OpTest):
def setUp(self):
self.op_type = "reduce_all"
self.python_api = reduce_all_wrapper
self.inputs = {'X': np.random.randint(0, 2, (5, 6, 10)).astype("bool")}
self.outputs = {'Out': self.inputs['X'].all()}
self.attrs = {'reduce_all': True}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAllFloatOp(OpTest):
def setUp(self):
self.op_type = "reduce_all"
self.python_api = reduce_all_wrapper
self.inputs = {'X': np.random.randint(0, 2, (5, 6, 10)).astype("float")}
self.outputs = {'Out': self.inputs['X'].all()}
self.attrs = {'reduce_all': True}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAllIntOp(OpTest):
def setUp(self):
self.op_type = "reduce_all"
self.python_api = reduce_all_wrapper
self.inputs = {'X': np.random.randint(0, 2, (5, 6, 10)).astype("int")}
self.outputs = {'Out': self.inputs['X'].all()}
self.attrs = {'reduce_all': True}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAllOp_ZeroDim(OpTest):
def setUp(self):
self.python_api = paddle.all
self.op_type = "reduce_all"
self.inputs = {'X': np.random.randint(0, 2, []).astype("bool")}
self.outputs = {'Out': self.inputs['X'].all()}
self.attrs = {'dim': []}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAll8DOp(OpTest):
def setUp(self):
self.op_type = "reduce_all"
self.python_api = paddle.all
self.inputs = {
'X': np.random.randint(0, 2, (2, 5, 3, 2, 2, 3, 4, 2)).astype(
"bool"
)
}
self.attrs = {'dim': (2, 3, 4)}
self.outputs = {'Out': self.inputs['X'].all(axis=self.attrs['dim'])}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAllOpWithDim(OpTest):
def setUp(self):
self.op_type = "reduce_all"
self.python_api = paddle.all
self.inputs = {'X': np.random.randint(0, 2, (5, 6, 10)).astype("bool")}
self.attrs = {'dim': (1,)}
self.outputs = {'Out': self.inputs['X'].all(axis=self.attrs['dim'])}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAll8DOpWithDim(OpTest):
def setUp(self):
self.op_type = "reduce_all"
self.python_api = paddle.all
self.inputs = {
'X': np.random.randint(0, 2, (2, 5, 3, 2, 2, 3, 4, 2)).astype(
"bool"
)
}
self.attrs = {'dim': (1, 3, 4)}
self.outputs = {'Out': self.inputs['X'].all(axis=self.attrs['dim'])}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAllOpWithKeepDim(OpTest):
def setUp(self):
self.op_type = "reduce_all"
self.python_api = paddle.all
self.inputs = {'X': np.random.randint(0, 2, (5, 6, 10)).astype("bool")}
self.attrs = {'dim': [1], 'keep_dim': True}
self.outputs = {
'Out': np.expand_dims(self.inputs['X'].all(axis=1), axis=1)
}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAll8DOpWithKeepDim(OpTest):
def setUp(self):
self.op_type = "reduce_all"
self.python_api = paddle.all
self.inputs = {
'X': np.random.randint(0, 2, (2, 5, 3, 2, 2, 3, 4, 2)).astype(
"bool"
)
}
self.attrs = {'dim': (5,), 'keep_dim': True}
self.outputs = {
'Out': np.expand_dims(
self.inputs['X'].all(axis=self.attrs['dim']), axis=5
)
}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAllComplex64Op(OpTest):
def setUp(self):
self.op_type = "reduce_all"
self.python_api = paddle.all
real_part = np.random.uniform(-1, 1, (2, 5, 3, 2, 2, 3, 4, 2))
imag_part = np.random.uniform(-1, 1, (2, 5, 3, 2, 2, 3, 4, 2))
self.inputs = {'X': (real_part + 1j * imag_part).astype("complex64")}
self.attrs = {'dim': (5,), 'keep_dim': True}
self.outputs = {
'Out': np.expand_dims(
self.inputs['X'].all(axis=self.attrs['dim']), axis=5
)
}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAllComplex64OpInf(TestAllComplex64Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.inf)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.inf)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex64")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAllComplex64OpNegInf(TestAllComplex64Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), -np.inf)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), -np.inf)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex64")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAllComplex64OpNan(TestAllComplex64Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.nan)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.nan)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex64")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAllComplex64OpZero(TestAllComplex64Op):
def setUp(self):
super().setUp()
real_part = np.zeros((2, 5, 3, 2, 2, 3, 4, 2))
imag_part = np.zeros((2, 5, 3, 2, 2, 3, 4, 2))
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex64")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAllComplex64OpMixed(TestAllComplex64Op):
def setUp(self):
super().setUp()
special_values = np.array(
[np.inf, -np.inf, np.nan, 0], dtype=np.float64
)
real_part = np.random.choice(special_values, (2, 5, 3, 2, 2, 3, 4, 2))
imag_part = np.random.choice(special_values, (2, 5, 3, 2, 2, 3, 4, 2))
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex64")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAllComplex128Op(OpTest):
def setUp(self):
self.op_type = "reduce_all"
self.python_api = paddle.all
real_part = np.random.uniform(-1, 1, (2, 5, 3, 2, 2, 3, 4, 2))
imag_part = np.random.uniform(-1, 1, (2, 5, 3, 2, 2, 3, 4, 2))
self.inputs = {'X': (real_part + 1j * imag_part).astype("complex128")}
self.attrs = {'dim': (5,), 'keep_dim': True}
self.outputs = {
'Out': np.expand_dims(
self.inputs['X'].all(axis=self.attrs['dim']), axis=5
)
}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAllComplex128OpInf(TestAllComplex128Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.inf)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.inf)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex128")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAllComplex128OpNegInf(TestAllComplex128Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), -np.inf)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), -np.inf)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex128")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAllComplex128OpNan(TestAllComplex128Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.nan)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.nan)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex128")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAllComplex128OpZero(TestAllComplex128Op):
def setUp(self):
super().setUp()
real_part = np.zeros((2, 5, 3, 2, 2, 3, 4, 2))
imag_part = np.zeros((2, 5, 3, 2, 2, 3, 4, 2))
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex128")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAllComplex128OpMixed(TestAllComplex128Op):
def setUp(self):
super().setUp()
special_values = np.array(
[np.inf, -np.inf, np.nan, 0], dtype=np.float64
)
real_part = np.random.choice(special_values, (2, 5, 3, 2, 2, 3, 4, 2))
imag_part = np.random.choice(special_values, (2, 5, 3, 2, 2, 3, 4, 2))
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex128")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAllOpError(unittest.TestCase):
def test_errors(self):
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
# The input type of reduce_all_op must be Variable.
input1 = 12
self.assertRaises(TypeError, paddle.all, input1)
def reduce_any_wrapper(x, axis=None, keepdim=False, reduce_all=True, name=None):
return paddle.any(x, axis, keepdim, name)
class TestAnyOp(OpTest):
def setUp(self):
self.op_type = "reduce_any"
self.prim_op_type = "comp"
self.python_api = reduce_any_wrapper
self.public_python_api = reduce_any_wrapper
self.inputs = {'X': np.random.randint(0, 2, (5, 6, 10)).astype("bool")}
self.outputs = {'Out': self.inputs['X'].any()}
self.attrs = {'reduce_all': True}
def test_check_output(self):
self.check_output(check_pir=True, check_prim_pir=True)
class TestAnyFloatOp(OpTest):
def setUp(self):
self.op_type = "reduce_any"
self.prim_op_type = "comp"
self.python_api = reduce_any_wrapper
self.public_python_api = reduce_any_wrapper
self.inputs = {'X': np.random.randint(0, 2, (5, 6, 10)).astype("float")}
self.outputs = {'Out': self.inputs['X'].any()}
self.attrs = {'reduce_all': True}
def test_check_output(self):
self.check_output(check_pir=True, check_prim_pir=True)
class TestAnyIntOp(OpTest):
def setUp(self):
self.op_type = "reduce_any"
self.prim_op_type = "comp"
self.python_api = reduce_any_wrapper
self.public_python_api = reduce_any_wrapper
self.inputs = {'X': np.random.randint(0, 2, (5, 6, 10)).astype("int")}
self.outputs = {'Out': self.inputs['X'].any()}
self.attrs = {'reduce_all': True}
def test_check_output(self):
self.check_output(check_pir=True, check_prim_pir=True)
class TestAnyComplex64Op(OpTest):
def setUp(self):
self.op_type = "reduce_any"
self.python_api = paddle.any
real_part = np.random.uniform(-1, 1, (2, 5, 3, 2, 2, 3, 4, 2))
imag_part = np.random.uniform(-1, 1, (2, 5, 3, 2, 2, 3, 4, 2))
self.inputs = {'X': (real_part + 1j * imag_part).astype("complex64")}
self.attrs = {'dim': (5,), 'keep_dim': True}
self.outputs = {
'Out': np.expand_dims(
self.inputs['X'].all(axis=self.attrs['dim']), axis=5
)
}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAnyComplex64OpInf(TestAnyComplex64Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.inf)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.inf)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex64")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAnyComplex64OpNegInf(TestAnyComplex64Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), -np.inf)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), -np.inf)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex64")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAnyComplex64OpNan(TestAnyComplex64Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.nan)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.nan)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex64")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAnyComplex64OpZero(TestAnyComplex64Op):
def setUp(self):
super().setUp()
real_part = np.zeros((2, 5, 3, 2, 2, 3, 4, 2))
imag_part = np.zeros((2, 5, 3, 2, 2, 3, 4, 2))
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex64")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAnyComplex128Op(OpTest):
def setUp(self):
self.op_type = "reduce_any"
self.python_api = paddle.any
real_part = np.random.uniform(-1, 1, (2, 5, 3, 2, 2, 3, 4, 2))
imag_part = np.random.uniform(-1, 1, (2, 5, 3, 2, 2, 3, 4, 2))
self.inputs = {'X': (real_part + 1j * imag_part).astype("complex128")}
self.attrs = {'dim': (5,), 'keep_dim': True}
self.outputs = {
'Out': np.expand_dims(
self.inputs['X'].all(axis=self.attrs['dim']), axis=5
)
}
def test_check_output(self):
self.check_output(check_pir=True)
class TestAnyComplex128OpInf(TestAnyComplex128Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.inf)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.inf)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex128")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAnyComplex128OpNegInf(TestAnyComplex128Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), -np.inf)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), -np.inf)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex128")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAnyComplex128OpNan(TestAnyComplex128Op):
def setUp(self):
super().setUp()
real_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.nan)
imag_part = np.full((2, 5, 3, 2, 2, 3, 4, 2), np.nan)
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex128")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAnyComplex128OpZero(TestAnyComplex128Op):
def setUp(self):
super().setUp()
real_part = np.zeros((2, 5, 3, 2, 2, 3, 4, 2))
imag_part = np.zeros((2, 5, 3, 2, 2, 3, 4, 2))
self.inputs['X'] = (real_part + 1j * imag_part).astype("complex128")
self.outputs['Out'] = np.expand_dims(
np.all(self.inputs['X'], axis=self.attrs['dim']), axis=5
)
class TestAnyOp_ZeroDim(OpTest):
def setUp(self):
self.op_type = "reduce_any"
self.prim_op_type = "comp"
self.python_api = paddle.any
self.public_python_api = paddle.any
self.inputs = {'X': np.random.randint(0, 2, []).astype("bool")}
self.outputs = {'Out': self.inputs['X'].any()}
self.attrs = {'dim': []}
def test_check_output(self):
self.check_output(check_pir=True, check_prim_pir=True)
class TestAny8DOp(OpTest):
def setUp(self):
self.op_type = "reduce_any"
self.prim_op_type = "comp"
self.python_api = paddle.any
self.public_python_api = paddle.any
self.inputs = {
'X': np.random.randint(0, 2, (2, 5, 3, 2, 2, 3, 4, 2)).astype(
"bool"
)
}
self.attrs = {'dim': (3, 5, 4)}
self.outputs = {'Out': self.inputs['X'].any(axis=self.attrs['dim'])}
def test_check_output(self):
self.check_output(check_pir=True, check_prim_pir=True)
class TestAnyOpWithDim(OpTest):
def setUp(self):
self.op_type = "reduce_any"
self.prim_op_type = "comp"
self.python_api = paddle.any
self.public_python_api = paddle.any
self.inputs = {'X': np.random.randint(0, 2, (5, 6, 10)).astype("bool")}
self.attrs = {'dim': [1]}
self.outputs = {'Out': self.inputs['X'].any(axis=1)}
def test_check_output(self):
self.check_output(check_pir=True, check_prim_pir=True)
class TestAny8DOpWithDim(OpTest):
def setUp(self):
self.op_type = "reduce_any"
self.prim_op_type = "comp"
self.python_api = paddle.any
self.public_python_api = paddle.any
self.inputs = {
'X': np.random.randint(0, 2, (2, 5, 3, 2, 2, 3, 4, 2)).astype(
"bool"
)
}
self.attrs = {'dim': (3, 6)}
self.outputs = {'Out': self.inputs['X'].any(axis=self.attrs['dim'])}
def test_check_output(self):
self.check_output(check_pir=True, check_prim_pir=True)
class TestAnyOpWithKeepDim(OpTest):
def setUp(self):
self.op_type = "reduce_any"
self.prim_op_type = "comp"
self.python_api = paddle.any
self.public_python_api = paddle.any
self.inputs = {'X': np.random.randint(0, 2, (5, 6, 10)).astype("bool")}
self.attrs = {'dim': (1,), 'keep_dim': True}
self.outputs = {
'Out': np.expand_dims(
self.inputs['X'].any(axis=self.attrs['dim']), axis=1
)
}
def test_check_output(self):
self.check_output(check_pir=True, check_prim_pir=True)
class TestAny8DOpWithKeepDim(OpTest):
def setUp(self):
self.op_type = "reduce_any"
self.prim_op_type = "comp"
self.python_api = paddle.any
self.public_python_api = paddle.any
self.inputs = {
'X': np.random.randint(0, 2, (2, 5, 3, 2, 2, 3, 4, 2)).astype(
"bool"
)
}
self.attrs = {'dim': (1,), 'keep_dim': True}
self.outputs = {
'Out': np.expand_dims(
self.inputs['X'].any(axis=self.attrs['dim']), axis=1
)
}
def test_check_output(self):
self.check_output(check_pir=True, check_prim_pir=True)
class TestAnyOpError(unittest.TestCase):
def test_errors(self):
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
# The input type of reduce_any_op must be Variable.
input1 = 12
self.assertRaises(TypeError, paddle.any, input1)
class Test1DReduce(OpTest):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.inputs = {'X': np.random.random(120).astype("float64")}
self.outputs = {'Out': self.inputs['X'].sum(axis=0)}
self.if_enable_cinn()
def if_enable_cinn(self):
pass
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False, check_prim_pir=True)
class TestReduceSum_ZeroDim(Test1DReduce):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.inputs = {'X': np.random.random(()).astype("float64")}
self.outputs = {'Out': self.inputs['X'].sum(axis=0)}
self.if_enable_cinn()
class Test2DReduce0(Test1DReduce):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.attrs = {'dim': [0]}
self.inputs = {'X': np.random.random((20, 10)).astype("float64")}
self.outputs = {'Out': self.inputs['X'].sum(axis=0)}
self.if_enable_cinn()
class Test2DReduce1(Test1DReduce):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.attrs = {'dim': [1]}
self.inputs = {'X': np.random.random((20, 10)).astype("float64")}
self.outputs = {
'Out': self.inputs['X'].sum(axis=tuple(self.attrs['dim']))
}
self.if_enable_cinn()
class Test3DReduce0(Test1DReduce):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.attrs = {'dim': [1]}
self.inputs = {'X': np.random.random((5, 6, 7)).astype("float64")}
self.outputs = {
'Out': self.inputs['X'].sum(axis=tuple(self.attrs['dim']))
}
self.if_enable_cinn()
class Test3DReduce1(Test1DReduce):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.attrs = {'dim': [2]}
self.inputs = {'X': np.random.random((5, 6, 7)).astype("float64")}
self.outputs = {
'Out': self.inputs['X'].sum(axis=tuple(self.attrs['dim']))
}
self.if_enable_cinn()
class Test3DReduce2(Test1DReduce):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.attrs = {'dim': [-2]}
self.inputs = {'X': np.random.random((5, 6, 7)).astype("float64")}
self.outputs = {
'Out': self.inputs['X'].sum(axis=tuple(self.attrs['dim']))
}
self.if_enable_cinn()
class Test3DReduce3(Test1DReduce):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.attrs = {'dim': [1, 2]}
self.inputs = {'X': np.random.random((5, 6, 7)).astype("float64")}
self.outputs = {
'Out': self.inputs['X'].sum(axis=tuple(self.attrs['dim']))
}
self.if_enable_cinn()
def reduce_sum_wrapper2(x, axis=[0], dtype=None, keepdim=False):
if paddle.in_dynamic_mode():
return paddle._C_ops.sum(x, axis, dtype, keepdim)
else:
if in_pir_mode():
return paddle._pir_ops.sum(x, axis, dtype, keepdim)
class Test8DReduce0(Test1DReduce):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = reduce_sum_wrapper2
self.attrs = {'dim': (4, 2, 3)}
self.inputs = {
'X': np.random.random((2, 5, 3, 2, 2, 3, 4, 2)).astype("float64")
}
self.outputs = {
'Out': self.inputs['X'].sum(axis=tuple(self.attrs['dim']))
}
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out')
class TestKeepDimReduce(Test1DReduce):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.inputs = {'X': np.random.random((5, 6, 10)).astype("float64")}
self.attrs = {'dim': [1], 'keep_dim': True}
self.outputs = {
'Out': self.inputs['X'].sum(
axis=tuple(self.attrs['dim']), keepdims=self.attrs['keep_dim']
)
}
self.if_enable_cinn()
class TestKeepDimReduceForEager(Test1DReduce):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = reduce_sum_wrapper2
self.inputs = {'X': np.random.random((5, 6, 10)).astype("float64")}
self.attrs = {'dim': [1], 'keep_dim': True}
self.outputs = {
'Out': self.inputs['X'].sum(
axis=tuple(self.attrs['dim']), keepdims=self.attrs['keep_dim']
)
}
def test_check_grad(self):
self.check_grad(['X'], 'Out')
class TestKeepDim8DReduce(Test1DReduce):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = reduce_sum_wrapper2
self.inputs = {
'X': np.random.random((2, 5, 3, 2, 2, 3, 4, 2)).astype("float64")
}
self.attrs = {'dim': (3, 4, 5), 'keep_dim': True}
self.outputs = {
'Out': self.inputs['X'].sum(
axis=tuple(self.attrs['dim']), keepdims=self.attrs['keep_dim']
)
}
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out')
@skip_check_grad_ci(
reason="reduce_max is discontinuous non-derivable function,"
" its gradient check is not supported by unittest framework."
)
class TestReduceMaxOpMultiAxes(OpTest):
"""Remove Max with subgradient from gradient check to confirm the success of CI."""
def setUp(self):
self.op_type = "reduce_max"
self.prim_op_type = "prim"
self.python_api = paddle.max
self.public_python_api = paddle.max
self.inputs = {'X': np.random.random((5, 6, 10)).astype("float64")}
self.attrs = {'dim': [-2, -1]}
self.outputs = {
'Out': self.inputs['X'].max(axis=tuple(self.attrs['dim']))
}
def test_check_output(self):
self.check_output(check_pir=True)
def test_check_grad(self):
# only composite op support gradient check of reduce_max
self.check_grad(
['X'],
'Out',
check_prim=False,
only_check_prim=True,
check_pir=True,
)
@skip_check_grad_ci(
reason="reduce_min is discontinuous non-derivable function,"
" its gradient check is not supported by unittest framework."
)
class TestReduceMinOpMultiAxes(OpTest):
"""Remove Min with subgradient from gradient check to confirm the success of CI."""
def setUp(self):
self.op_type = "reduce_min"
self.python_api = paddle.min
self.inputs = {'X': np.random.random((5, 6, 10)).astype("float64")}
self.attrs = {'dim': [1, 2]}
self.outputs = {
'Out': self.inputs['X'].min(axis=tuple(self.attrs['dim']))
}
def test_check_output(self):
self.check_output()
class TestKeepDimReduceSumMultiAxes(OpTest):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.inputs = {'X': np.random.random((5, 6, 10)).astype("float64")}
self.attrs = {'dim': [-2, -1], 'keep_dim': True}
self.outputs = {
'Out': self.inputs['X'].sum(
axis=tuple(self.attrs['dim']), keepdims=True
)
}
self.if_enable_cinn()
def if_enable_cinn(self):
pass
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False, check_prim_pir=True)
class TestKeepDimReduceSumMultiAxesForEager(OpTest):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = reduce_sum_wrapper2
self.inputs = {'X': np.random.random((5, 6, 10)).astype("float64")}
self.attrs = {'dim': [-2, -1], 'keep_dim': True}
self.outputs = {
'Out': self.inputs['X'].sum(
axis=tuple(self.attrs['dim']), keepdims=True
)
}
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out')
class TestReduceSumWithDimOne(OpTest):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.inputs = {'X': np.random.random((100, 1, 1)).astype("float64")}
self.attrs = {'dim': [1, 2], 'keep_dim': True}
self.outputs = {
'Out': self.inputs['X'].sum(
axis=tuple(self.attrs['dim']), keepdims=True
)
}
self.if_enable_cinn()
def if_enable_cinn(self):
pass
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False, check_prim_pir=True)
class TestReduceSumWithDimOneForEager(OpTest):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = reduce_sum_wrapper2
self.inputs = {'X': np.random.random((100, 1, 1)).astype("float64")}
self.attrs = {'dim': [1, 2], 'keep_dim': True}
self.outputs = {
'Out': self.inputs['X'].sum(
axis=tuple(self.attrs['dim']), keepdims=True
)
}
self.enable_cinn = True
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out')
class TestReduceSumWithNumelOne(OpTest):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.inputs = {'X': np.random.random((100, 1)).astype("float64")}
self.attrs = {'dim': [1], 'keep_dim': False}
self.outputs = {
'Out': self.inputs['X'].sum(
axis=tuple(self.attrs['dim']), keepdims=False
)
}
self.if_enable_cinn()
def if_enable_cinn(self):
pass
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False)
def reduce_sum_wrapper(
x, axis=None, keepdim=False, reduce_all=True, out_dtype=None, name=None
):
return paddle.sum(x, axis, out_dtype, keepdim, name)
class TestReduceAll(OpTest):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = reduce_sum_wrapper
self.public_python_api = reduce_sum_wrapper
self.prim_op_type = "prim"
self.inputs = {'X': np.random.random((100, 1, 1)).astype("float64")}
self.attrs = {'reduce_all': True, 'keep_dim': False}
self.outputs = {'Out': self.inputs['X'].sum()}
self.if_enable_cinn()
def if_enable_cinn(self):
pass
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False, check_prim_pir=True)
class TestReduceAllFp32(OpTest):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = reduce_sum_wrapper
self.public_python_api = reduce_sum_wrapper
self.prim_op_type = "prim"
self.inputs = {'X': np.random.random((100, 1, 1)).astype("float32")}
self.attrs = {'reduce_all': True, 'keep_dim': False}
self.outputs = {'Out': self.inputs['X'].sum()}
self.if_enable_cinn()
def if_enable_cinn(self):
pass
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False, check_prim_pir=True)
class Test1DReduceWithAxes1(OpTest):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.inputs = {'X': np.random.random(100).astype("float64")}
self.attrs = {'dim': [0], 'keep_dim': False}
self.outputs = {'Out': self.inputs['X'].sum(axis=0)}
self.if_enable_cinn()
def if_enable_cinn(self):
pass
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False, check_prim_pir=True)
def reduce_sum_wrapper_fp64(
x, axis=None, keepdim=False, reduce_all=True, out_dtype=None, name=None
):
return paddle.sum(x, axis, 'float64', keepdim, name)
class TestReduceWithDtype(OpTest):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = reduce_sum_wrapper_fp64
self.public_python_api = reduce_sum_wrapper_fp64
self.prim_op_type = "prim"
self.inputs = {'X': np.random.random((6, 2, 10)).astype("float64")}
self.outputs = {'Out': self.inputs['X'].sum().astype('float64')}
self.attrs = {'reduce_all': True}
self.attrs.update(
{
'in_dtype': paddle.float32,
'out_dtype': paddle.float64,
}
)
self.if_enable_cinn()
def if_enable_cinn(self):
pass
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False, check_prim_pir=True)
class TestReduceWithDtype1(TestReduceWithDtype):
def setUp(self):
self.op_type = "reduce_sum"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.prim_op_type = "prim"
self.inputs = {'X': np.random.random((6, 2, 10)).astype("float64")}
self.outputs = {'Out': self.inputs['X'].sum(axis=1)}
self.attrs = {'dim': [1]}
self.attrs.update(
{
'in_dtype': paddle.float32,
'out_dtype': paddle.float64,
}
)
# cinn op_mapper not support in_dtype/out_dtype attr
self.enable_cinn = False
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False, check_prim_pir=True)
class TestReduceWithDtype2(TestReduceWithDtype):
def setUp(self):
self.op_type = "reduce_sum"
self.prim_op_type = "prim"
self.python_api = paddle.sum
self.public_python_api = paddle.sum
self.inputs = {'X': np.random.random((6, 2, 10)).astype("float64")}
self.outputs = {'Out': self.inputs['X'].sum(axis=1, keepdims=True)}
self.attrs = {'dim': [1], 'keep_dim': True}
self.attrs.update(
{
'in_dtype': paddle.float32,
'out_dtype': paddle.float64,
}
)
# cinn op_mapper not support in_dtype/out_dtype attr
self.enable_cinn = False
def test_check_output(self):
self.check_output()
def test_check_grad(self):
self.check_grad(['X'], 'Out', check_prim=False, check_prim_pir=True)
class TestReduceSumOpError(unittest.TestCase):
def test_errors1(self):
with (
static_guard(),
paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
),
):
# The input type of reduce_sum_op must be Variable.
x1 = base.create_lod_tensor(
np.array([[-1]]), [[1]], base.CPUPlace()
)
self.assertRaises(TypeError, paddle.sum, x1)
# The input dtype of reduce_sum_op must be float32 or float64 or int32 or int64.
class API_TestSumOp(unittest.TestCase):
def run_static(
self, shape, x_dtype, attr_axis, attr_dtype=None, np_axis=None
):
if np_axis is None:
np_axis = attr_axis
places = get_places()
for place in places:
with base.program_guard(base.Program(), base.Program()):
data = paddle.static.data("data", shape=shape, dtype=x_dtype)
result_sum = paddle.sum(
x=data, axis=attr_axis, dtype=attr_dtype
)
exe = base.Executor(place)
input_data = np.random.rand(*shape).astype(x_dtype)
(res,) = exe.run(
feed={"data": input_data}, fetch_list=[result_sum]
)
np.testing.assert_allclose(
res,
np.sum(input_data.astype(attr_dtype), axis=np_axis),
rtol=1e-05,
)
def test_static(self):
shape = [10, 10]
axis = 1
self.run_static(shape, "bool", axis, attr_dtype=None)
self.run_static(shape, "bool", axis, attr_dtype="int32")
self.run_static(shape, "bool", axis, attr_dtype="int64")
self.run_static(shape, "bool", axis, attr_dtype="float16")
self.run_static(shape, "int32", axis, attr_dtype=None)
self.run_static(shape, "int32", axis, attr_dtype="int32")
self.run_static(shape, "int32", axis, attr_dtype="int64")
self.run_static(shape, "int32", axis, attr_dtype="float64")
self.run_static(shape, "int64", axis, attr_dtype=None)
self.run_static(shape, "int64", axis, attr_dtype="int64")
self.run_static(shape, "int64", axis, attr_dtype="int32")
self.run_static(shape, "float32", axis, attr_dtype=None)
self.run_static(shape, "float32", axis, attr_dtype="float32")
self.run_static(shape, "float32", axis, attr_dtype="float64")
self.run_static(shape, "float32", axis, attr_dtype="int64")
self.run_static(shape, "float64", axis, attr_dtype=None)
self.run_static(shape, "float64", axis, attr_dtype="float32")
self.run_static(shape, "float64", axis, attr_dtype="float64")
shape = [5, 5, 5]
self.run_static(shape, "int32", (0, 1), attr_dtype="int32")
self.run_static(
shape, "int32", (), attr_dtype="int32", np_axis=(0, 1, 2)
)
def test_dygraph(self):
np_x = np.random.random([2, 3, 4]).astype('int32')
with base.dygraph.guard():
x = paddle.to_tensor(np_x)
out0 = paddle.sum(x).numpy()
out1 = paddle.sum(x, axis=0).numpy()
out2 = paddle.sum(x, axis=(0, 1)).numpy()
out3 = paddle.sum(x, axis=(0, 1, 2)).numpy()
self.assertTrue((out0 == np.sum(np_x, axis=(0, 1, 2))).all())
self.assertTrue((out1 == np.sum(np_x, axis=0)).all())
self.assertTrue((out2 == np.sum(np_x, axis=(0, 1))).all())
self.assertTrue((out3 == np.sum(np_x, axis=(0, 1, 2))).all())
class TestAllAPI(unittest.TestCase):
def setUp(self):
np.random.seed(123)
paddle.enable_static()
self.places = get_places()
def check_static_result(self, place):
main = paddle.static.Program()
startup = paddle.static.Program()
with base.program_guard(main, startup):
input = paddle.static.data(name="input", shape=[4, 4], dtype="bool")
result = paddle.all(x=input)
input_np = np.random.randint(0, 2, [4, 4]).astype("bool")
exe = base.Executor(place)
fetches = exe.run(
main,
feed={"input": input_np},
fetch_list=[result],
)
self.assertTrue((fetches[0] == np.all(input_np)).all())
def check_static_float_result(self, place):
main = paddle.static.Program()
startup = paddle.static.Program()
with base.program_guard(main, startup):
input = paddle.static.data(
name="input", shape=[4, 4], dtype="float"
)
result = paddle.all(x=input)
input_np = np.random.randint(0, 2, [4, 4]).astype("float")
exe = base.Executor(place)
fetches = exe.run(
main,
feed={"input": input_np},
fetch_list=[result],
)
self.assertTrue((fetches[0] == np.all(input_np)).all())
def check_static_int_result(self, place):
main = paddle.static.Program()
startup = paddle.static.Program()
with base.program_guard(main, startup):
input = paddle.static.data(name="input", shape=[4, 4], dtype="int")
result = paddle.all(x=input)
input_np = np.random.randint(0, 2, [4, 4]).astype("int")
exe = base.Executor(place)
fetches = exe.run(
main,
feed={"input": input_np},
fetch_list=[result],
)
self.assertTrue((fetches[0] == np.all(input_np)).all())
def test_static(self):
for place in self.places:
self.check_static_result(place=place)
self.check_static_float_result(place=place)
self.check_static_int_result(place=place)
def test_dygraph(self):
paddle.disable_static()
for place in self.places:
with base.dygraph.guard(place):
np_x = np.random.randint(0, 2, (12, 10)).astype(np.bool_)
x = paddle.assign(np_x)
x = paddle.cast(x, 'bool')
out1 = paddle.all(x)
np_out1 = out1.numpy()
expect_res1 = np.all(np_x)
self.assertTrue((np_out1 == expect_res1).all())
out2 = paddle.all(x, axis=0)
np_out2 = out2.numpy()
expect_res2 = np.all(np_x, axis=0)
self.assertTrue((np_out2 == expect_res2).all())
out3 = paddle.all(x, axis=-1)
np_out3 = out3.numpy()
expect_res3 = np.all(np_x, axis=-1)
self.assertTrue((np_out3 == expect_res3).all())
out4 = paddle.all(x, axis=1, keepdim=True)
np_out4 = out4.numpy()
expect_res4 = np.all(np_x, axis=1, keepdims=True)
self.assertTrue((np_out4 == expect_res4).all())
x = paddle.cast(x, 'float')
out5 = paddle.all(x)
np_out5 = out5.numpy()
expect_res5 = np.all(np_x)
self.assertTrue((np_out5 == expect_res5).all())
x = paddle.cast(x, 'int')
out6 = paddle.all(x)
np_out6 = out6.numpy()
expect_res6 = np.all(np_x)
self.assertTrue((np_out6 == expect_res6).all())
paddle.enable_static()
class TestAllAPI_Compatibility(unittest.TestCase):
def setUp(self):
np.random.seed(123)
paddle.enable_static()
self.places = get_places()
self.shape = [5, 6]
self.dtype = 'bool'
self.init_data()
def init_data(self):
self.np_input = np.random.randint(0, 8, self.shape).astype(self.dtype)
def test_dygraph_Compatibility(self):
paddle.disable_static()
x = paddle.to_tensor(self.np_input)
paddle_dygraph_out = []
# Position args (args)
out1 = paddle.all(x, 1, True)
paddle_dygraph_out.append(out1)
# Keywords args (kwargs) for paddle
out2 = paddle.all(x=x, axis=1, keepdim=True)
paddle_dygraph_out.append(out2)
# Keywords args for torch
out3 = paddle.all(input=x, dim=1, keepdim=True)
paddle_dygraph_out.append(out3)
# Combined args and kwargs
out4 = paddle.all(x, dim=1, keepdim=True)
paddle_dygraph_out.append(out4)
# Tensor method args
out5 = x.all(1, True)
paddle_dygraph_out.append(out5)
# Tensor method kwargs
out6 = x.all(dim=1, keepdim=True)
paddle_dygraph_out.append(out6)
# Test out
out7 = paddle.empty([])
paddle.all(x, 1, True, out=out7)
paddle_dygraph_out.append(out7)
# Numpy reference out
ref_out = np.all(self.np_input, 1, keepdims=True)
# Check
for out in paddle_dygraph_out:
np.testing.assert_allclose(ref_out, out.numpy())
paddle.enable_static()
def test_static_Compatibility(self):
main = paddle.static.Program()
startup = paddle.static.Program()
with base.program_guard(main, startup):
x = paddle.static.data(name="x", shape=self.shape, dtype=self.dtype)
# Position args (args)
out1 = paddle.all(x, 1, True)
# Keywords args (kwargs) for paddle
out2 = paddle.all(x=x, axis=1, keepdim=True)
# Keywords args for torch
out3 = paddle.all(input=x, dim=1, keepdim=True)
# Combined args and kwargs
out4 = paddle.all(x, dim=1, keepdim=True)
# Tensor method args
out5 = x.all(1, True)
# Tensor method kwargs
out6 = x.all(dim=1, keepdim=True)
# Do not support out in static
# out7 = paddle.empty([])
# paddle.all(x, 1, True, out=out7)
exe = base.Executor(paddle.CPUPlace())
fetches = exe.run(
main,
feed={"x": self.np_input},
fetch_list=[out1, out2, out3, out4, out5, out6],
)
ref_out = np.all(self.np_input, 1, keepdims=True)
for out in fetches:
np.testing.assert_allclose(out, ref_out)
class TestAnyAPI(unittest.TestCase):
def setUp(self):
np.random.seed(123)
paddle.enable_static()
self.places = get_places()
def check_static_result(self, place):
main = paddle.static.Program()
startup = paddle.static.Program()
with base.program_guard(main, startup):
input = paddle.static.data(name="input", shape=[4, 4], dtype="bool")
result = paddle.any(x=input)
input_np = np.random.randint(0, 2, [4, 4]).astype("bool")
exe = base.Executor(place)
fetches = exe.run(
main,
feed={"input": input_np},
fetch_list=[result],
)
self.assertTrue((fetches[0] == np.any(input_np)).all())
def check_static_float_result(self, place):
main = paddle.static.Program()
startup = paddle.static.Program()
with base.program_guard(main, startup):
input = paddle.static.data(
name="input", shape=[4, 4], dtype="float"
)
result = paddle.any(x=input)
input_np = np.random.randint(0, 2, [4, 4]).astype("float")
exe = base.Executor(place)
fetches = exe.run(
main,
feed={"input": input_np},
fetch_list=[result],
)
self.assertTrue((fetches[0] == np.any(input_np)).all())
def check_static_int_result(self, place):
main = paddle.static.Program()
startup = paddle.static.Program()
with base.program_guard(main, startup):
input = paddle.static.data(name="input", shape=[4, 4], dtype="int")
result = paddle.any(x=input)
input_np = np.random.randint(0, 2, [4, 4]).astype("int")
exe = base.Executor(place)
fetches = exe.run(
main,
feed={"input": input_np},
fetch_list=[result],
)
self.assertTrue((fetches[0] == np.any(input_np)).all())
def test_static(self):
for place in self.places:
self.check_static_result(place=place)
self.check_static_float_result(place=place)
self.check_static_int_result(place=place)
def test_dygraph(self):
paddle.disable_static()
for place in self.places:
with base.dygraph.guard(place):
np_x = np.random.randint(0, 2, (12, 10)).astype(np.bool_)
x = paddle.assign(np_x)
x = paddle.cast(x, 'bool')
out1 = paddle.any(x)
np_out1 = out1.numpy()
expect_res1 = np.any(np_x)
self.assertTrue((np_out1 == expect_res1).all())
out2 = paddle.any(x, axis=0)
np_out2 = out2.numpy()
expect_res2 = np.any(np_x, axis=0)
self.assertTrue((np_out2 == expect_res2).all())
out3 = paddle.any(x, axis=-1)
np_out3 = out3.numpy()
expect_res3 = np.any(np_x, axis=-1)
self.assertTrue((np_out3 == expect_res3).all())
out4 = paddle.any(x, axis=1, keepdim=True)
np_out4 = out4.numpy()
expect_res4 = np.any(np_x, axis=1, keepdims=True)
self.assertTrue((np_out4 == expect_res4).all())
np_x = np.random.randint(0, 2, (12, 10)).astype(np.float32)
x = paddle.assign(np_x)
x = paddle.cast(x, 'float32')
out5 = paddle.any(x)
np_out5 = out5.numpy()
expect_res5 = np.any(np_x)
self.assertTrue((np_out5 == expect_res5).all())
x = paddle.cast(x, 'int')
out6 = paddle.any(x)
np_out6 = out6.numpy()
expect_res6 = np.any(np_x)
self.assertTrue((np_out6 == expect_res6).all())
paddle.enable_static()
class TestAllZero(unittest.TestCase):
def setUp(self):
np.random.seed(123)
self.shape = [1, 0, 2]
self.dtypes = [
"bool",
"float32",
"float64",
"int32",
"complex64",
"complex128",
]
self.places = get_places()
def calculate_expected_result(self, x_np, axis, keepdim):
expected_result = np.all(x_np, axis=axis, keepdims=keepdim)
return expected_result
def check_result(
self, static_result, expected_result, axis, keepdim, dtype, place
):
self.assertTrue(
(static_result == expected_result).all(),
f"Static Mode - Shape: {self.shape}, Axis: {axis}, Keepdim: {keepdim}, Dtype: {dtype}, Place: {place}",
)
def _test_static(self, place, axis, keepdim, dtype):
with (
static_guard(),
base.program_guard(
paddle.static.Program(), paddle.static.Program()
),
):
input = paddle.static.data(name="x", shape=self.shape, dtype=dtype)
result = paddle.all(x=input, axis=axis, keepdim=keepdim)
x_np = np.zeros(self.shape, dtype=dtype)
exe = base.Executor(place)
fetches = exe.run(
feed={"x": x_np},
fetch_list=[result],
)
expected_result = self.calculate_expected_result(
x_np, axis, keepdim
)
self.check_result(
fetches[0], expected_result, axis, keepdim, dtype, place
)
def _test_dygraph(self, place, axis, keepdim, dtype):
with dygraph_guard():
x_np = np.zeros(self.shape, dtype=dtype)
x = paddle.to_tensor(x_np)
dygraph_result = paddle.all(x, axis=axis, keepdim=keepdim).numpy()
expected_result = self.calculate_expected_result(
x_np, axis, keepdim
)
self.check_result(
dygraph_result, expected_result, axis, keepdim, dtype, place
)
def _test_all(self, place, axis, keepdim, dtype):
self._test_dygraph(place, axis, keepdim, dtype)
self._test_static(place, axis, keepdim, dtype)
def test_zero_size(self):
axes_options = [
None,
0,
1,
2,
-1,
-2,
(),
(0, 1),
(0, 2),
(1, 2),
(-1, -2),
]
keepdims_options = [True, False]
for place in self.places:
for dtype in self.dtypes:
for axis in axes_options:
for keepdim in keepdims_options:
self._test_all(place, axis, keepdim, dtype)
class TestAnyZero(unittest.TestCase):
def setUp(self):
np.random.seed(123)
self.shape = [1, 0, 2]
self.dtypes = [
"bool",
"float32",
"float64",
"int32",
"complex64",
"complex128",
]
self.places = get_places()
def calculate_expected_result(self, x_np, axis, keepdim):
expected_result = np.any(x_np, axis=axis, keepdims=keepdim)
return expected_result
def check_result(
self, static_result, expected_result, axis, keepdim, dtype, place
):
self.assertTrue(
(static_result == expected_result).all(),
f"Static Mode - Shape: {self.shape}, Axis: {axis}, Keepdim: {keepdim}, Dtype: {dtype}, Place: {place}",
)
def _test_static(self, place, axis, keepdim, dtype):
with (
static_guard(),
base.program_guard(
paddle.static.Program(), paddle.static.Program()
),
):
input = paddle.static.data(name="x", shape=self.shape, dtype=dtype)
result = paddle.any(x=input, axis=axis, keepdim=keepdim)
x_np = np.zeros(self.shape, dtype=dtype)
exe = base.Executor(place)
fetches = exe.run(
feed={"x": x_np},
fetch_list=[result],
)
expected_result = self.calculate_expected_result(
x_np, axis, keepdim
)
self.check_result(
fetches[0], expected_result, axis, keepdim, dtype, place
)
def _test_dygraph(self, place, axis, keepdim, dtype):
with dygraph_guard():
x_np = np.zeros(self.shape, dtype=dtype)
x = paddle.to_tensor(x_np)
dygraph_result = paddle.any(x, axis=axis, keepdim=keepdim).numpy()
expected_result = self.calculate_expected_result(
x_np, axis, keepdim
)
self.check_result(
dygraph_result, expected_result, axis, keepdim, dtype, place
)
def _test_any(self, place, axis, keepdim, dtype):
self._test_dygraph(place, axis, keepdim, dtype)
self._test_static(place, axis, keepdim, dtype)
def test_zero_size(self):
axes_options = [
None,
0,
1,
2,
-1,
-2,
(),
(0, 1),
(0, 2),
(1, 2),
(-1, -2),
]
keepdims_options = [True, False]
for place in self.places:
for dtype in self.dtypes:
for axis in axes_options:
for keepdim in keepdims_options:
self._test_any(place, axis, keepdim, dtype)
class TestAnyCompatibility(unittest.TestCase):
def setUp(self):
self.places = [paddle.CPUPlace()]
if paddle.base.core.is_compiled_with_cuda():
self.places.append(get_device_place())
self.func = paddle.any
self.init_data()
self.init_case()
def init_data(self):
self.shape = [5, 6]
self.dtype = 'float32'
self.axis = 1
self.np_input = np.random.randint(0, 2, self.shape).astype(self.dtype)
self.np_out = np.any(self.np_input, self.axis, keepdims=True)
def init_case(self):
params = [['x', 'input'], ['axis', 'dim']] # param1 # param2
# Generate all valid combinations
def generate_cases(param_groups, case_list):
from itertools import product
for combo in product(*[[None, *names] for names in param_groups]):
args = ['pos' if p is None else 'kw' for p in combo]
if args == sorted(args, key=lambda x: x != 'pos'):
case_list.append(combo)
# paddle.chunk()
self.test_cases = []
generate_cases(params, self.test_cases)
# x.chunk()
self.tensor_test_cases = []
generate_cases(params[1:], self.tensor_test_cases)
def _build_args_kwargs(self, param_names, params):
args = []
kwargs = {}
for name, param in zip(param_names, params):
if name is None:
args.append(param)
else:
kwargs[name] = param
kwargs['keepdim'] = True
return args, kwargs
def test_dygraph_compatibility(self):
with dygraph_guard():
for place in self.places:
paddle.device.set_device(place)
x = paddle.to_tensor(self.np_input)
# paddle.
for param_names in self.test_cases:
args, kwargs = self._build_args_kwargs(
param_names, (x, self.axis)
)
for out_flag in [False, True]:
if out_flag:
kwargs['out'] = paddle.empty([])
self.func(*args, **kwargs)
out = kwargs["out"]
else:
out = self.func(*args, **kwargs)
np.testing.assert_allclose(
self.np_out, out.numpy(), rtol=1e-10
)
# paddle.Tensor.
for param_names in self.tensor_test_cases:
args, kwargs = self._build_args_kwargs(
param_names, (self.axis,)
)
out = x.any(*args, **kwargs)
np.testing.assert_allclose(
self.np_out, out.numpy(), rtol=1e-10
)
def test_dygraph_out(self):
def run_any(test_type):
x = paddle.to_tensor(self.np_input)
x.stop_gradient = False
out = (
paddle.zeros(self.np_out.shape)
if test_type in ["with_out", "both"]
else None
)
if test_type == "return":
out = paddle.any(x, axis=self.axis, keepdim=True)
elif test_type == "with_out":
paddle.any(x, axis=self.axis, keepdim=True, out=out)
elif test_type == "both":
out = paddle.any(x, axis=self.axis, keepdim=True, out=out)
else:
raise ValueError(f"Invalid test_mode: {test_type}")
expected = paddle._C_ops.any(x, self.axis, True)
np.testing.assert_array_equal(out.numpy(), expected.numpy())
loss = out.sum().astype('float32')
loss.backward()
return out, x.grad
def assert_outputs_equal(outputs, rtol: float = 1e-10):
for out in outputs[1:]:
np.testing.assert_allclose(
outputs[0].numpy(), out.numpy(), rtol=rtol
)
with dygraph_guard():
for place in self.places:
paddle.device.set_device(place)
out1, grad1 = run_any("return")
out2, grad2 = run_any("with_out")
out3, grad3 = run_any("both")
assert_outputs_equal([out1, out2, out3])
if (
grad1 is not None
and grad2 is not None
and grad3 is not None
):
assert_outputs_equal([grad1, grad2, grad3])
def test_static_compatibility(self):
with static_guard():
for place in self.places:
main = paddle.static.Program()
startup = paddle.static.Program()
with base.program_guard(main, startup):
x = paddle.static.data(
name="x", shape=self.shape, dtype=self.dtype
)
# paddle.
for param_names in self.test_cases:
args, kwargs = self._build_args_kwargs(
param_names, (x, self.axis)
)
out = self.func(*args, **kwargs)
exe = base.Executor(place)
fetches = exe.run(
main,
feed={"x": self.np_input},
fetch_list=[out],
)
np.testing.assert_allclose(
self.np_out, fetches[0], rtol=1e-10
)
# paddle.Tensor.
for param_names in self.tensor_test_cases:
args, kwargs = self._build_args_kwargs(
param_names, (self.axis,)
)
out = x.any(*args, **kwargs)
exe = base.Executor(place)
fetches = exe.run(
main,
feed={"x": self.np_input},
fetch_list=[out],
)
np.testing.assert_allclose(
self.np_out, fetches[0], rtol=1e-10
)
# Dimension exceeds int32 range.
class TestSumOpIndexInt32OverflowCase0(unittest.TestCase):
def setUp(self):
self.shape = [2147483678]
self.axis = 0
self.input_dtype = 'float32'
self.test_dtypes = [np.float32]
def test_dygraph(self):
with dygraph_guard():
x_paddle = paddle.ones(shape=self.shape, dtype=self.input_dtype)
for dtype_input in self.test_dtypes:
numpy_result = np.sum(
x_paddle.numpy(),
axis=self.axis,
dtype=np.dtype(dtype_input),
keepdims=False,
)
# paddle test case
paddle_result0 = paddle.sum(x_paddle, self.axis, dtype_input)
np.testing.assert_allclose(
paddle_result0, numpy_result, rtol=1e-05
)
# Index exceeds int32 range.
class TestSumOpIndexInt32OverflowCase1(unittest.TestCase):
def setUp(self):
self.shape = [1073741830]
self.axis = 0
self.input_dtype = 'float32'
self.test_dtypes = [np.float32]
def test_dygraph(self):
with dygraph_guard():
x_paddle = paddle.ones(shape=self.shape, dtype=self.input_dtype)
for dtype_input in self.test_dtypes:
numpy_result = np.sum(
x_paddle.numpy(),
axis=self.axis,
dtype=np.dtype(dtype_input),
keepdims=False,
)
# paddle test case
paddle_result0 = paddle.sum(x_paddle, self.axis, dtype_input)
np.testing.assert_allclose(
paddle_result0, numpy_result, rtol=1e-05
)
if __name__ == '__main__':
paddle.enable_static()
unittest.main()
Reference in New Issue View Git Blame Copy Permalink