Files
paddlepaddle--paddle/test/legacy_test/test_elementwise_heaviside_op.py
2026-07-13 12:40:42 +08:00

608 lines
22 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 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 unittest
import numpy as np
from op_test import (
OpTest,
convert_float_to_uint16,
get_device_place,
is_custom_device,
)
import paddle
from paddle.base import core
def Heaviside_grad(x, y, dout, astype="float16", is_bfloat16=False):
tmp = np.zeros(x.shape).astype(astype)
dx = np.multiply(tmp, dout)
dy = np.multiply(np.equal(x, 0), dout).astype(astype)
if is_bfloat16:
dx = convert_float_to_uint16(dx)
dy = convert_float_to_uint16(dy)
return dx, dy
class TestElementwiseOp(OpTest):
def setUp(self):
self.op_type = "elementwise_heaviside"
x = np.random.random((13, 17)).astype("float64")
y = np.random.random((13, 17)).astype("float64")
self.python_api = paddle.heaviside
self.prim_op_type = "comp"
self.public_python_api = paddle.heaviside
self.inputs = {'X': x, 'Y': y}
self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])}
def test_check_output(self):
self.check_output(
check_pir=True, check_prim_pir=True, check_symbol_infer=False
)
def test_check_grad_normal(self):
self.check_grad(['X', 'Y'], 'Out', check_pir=True, check_prim_pir=True)
def test_check_grad_ignore_x(self):
self.check_grad(
['Y'],
'Out',
no_grad_set=set("X"),
check_pir=True,
check_prim_pir=True,
)
def test_check_grad_ignore_y(self):
self.check_grad(
['X'],
'Out',
no_grad_set=set('Y'),
check_pir=True,
check_prim_pir=True,
)
class TestHeavisideBroadcast(unittest.TestCase):
def setUp(self):
self.input_1 = np.random.rand(2, 100, 13, 17).astype("float32")
self.input_2 = np.random.rand(100, 13, 17).astype("float32")
self.input_3 = np.random.rand(100, 13, 1).astype("float32")
self.input_4 = np.random.rand(13, 17).astype("float32")
self.input_5 = np.random.rand(1).astype("float32")
self.np_expected1 = np.heaviside(self.input_1, self.input_2)
self.np_expected2 = np.heaviside(self.input_2, self.input_3)
self.np_expected3 = np.heaviside(self.input_2, self.input_4)
self.np_expected4 = np.heaviside(self.input_4, self.input_5)
def test_broadcast(self):
paddle.disable_static()
self.tensor_1 = paddle.to_tensor(self.input_1)
self.tensor_2 = paddle.to_tensor(self.input_2)
self.tensor_3 = paddle.to_tensor(self.input_3)
self.tensor_4 = paddle.to_tensor(self.input_4)
self.tensor_5 = paddle.to_tensor(self.input_5)
res = paddle.heaviside(self.tensor_1, self.tensor_2)
res = res.numpy()
np.testing.assert_allclose(res, self.np_expected1, rtol=1e-05)
res = paddle.heaviside(self.tensor_2, self.tensor_3)
res = res.numpy()
np.testing.assert_allclose(res, self.np_expected2, rtol=1e-05)
res = paddle.heaviside(self.tensor_2, self.tensor_4)
res = res.numpy()
np.testing.assert_allclose(res, self.np_expected3, rtol=1e-05)
res = paddle.heaviside(self.tensor_4, self.tensor_5)
res = res.numpy()
np.testing.assert_allclose(res, self.np_expected4, rtol=1e-05)
class TestHeavisideAPI_float64(unittest.TestCase):
def setUp(self):
self.x_np = np.random.random((13, 17)).astype("float64")
self.y_np = np.random.random((13, 17)).astype("float64")
self.out_np = np.heaviside(self.x_np, self.y_np)
self.dtype = "float64"
def test_static(self):
for use_cuda in (
[False, True]
if (paddle.device.is_compiled_with_cuda() or is_custom_device())
else [False]
):
place = get_device_place() if use_cuda else paddle.CPUPlace()
paddle.enable_static()
prog = paddle.static.Program()
with paddle.static.program_guard(prog):
x = paddle.static.data(
name=f"x_{self.dtype}", shape=[13, 17], dtype=self.dtype
)
y = paddle.static.data(
name=f"y_{self.dtype}", shape=[13, 17], dtype=self.dtype
)
out = paddle.heaviside(x, y)
exe = paddle.static.Executor(place=place)
(res,) = exe.run(
prog,
feed={
f"x_{self.dtype}": self.x_np,
f"y_{self.dtype}": self.y_np,
},
fetch_list=out,
use_prune=True,
)
np.testing.assert_allclose(res, self.out_np, rtol=1e-05)
def test_dygraph(self):
for use_cuda in (
[False, True]
if (paddle.device.is_compiled_with_cuda() or is_custom_device())
else [False]
):
place = get_device_place() if use_cuda else paddle.CPUPlace()
paddle.disable_static(place=place)
result = paddle.heaviside(
paddle.to_tensor(self.x_np), paddle.to_tensor(self.y_np)
)
np.testing.assert_allclose(result.numpy(), self.out_np, rtol=1e-05)
class TestHeavisideAPI_float32(TestHeavisideAPI_float64):
def setUp(self):
self.x_np = np.random.random((13, 17)).astype("float32")
self.y_np = np.random.random((13, 17)).astype("float32")
self.out_np = np.heaviside(self.x_np, self.y_np)
self.dtype = "float32"
class TestHeavisideAPI_int64(TestHeavisideAPI_float64):
def setUp(self):
self.x_np = np.random.random((13, 17)).astype("int64")
self.y_np = np.random.random((13, 17)).astype("int64")
self.out_np = np.heaviside(self.x_np, self.y_np)
self.dtype = "int64"
class TestHeavisideAPI_int32(TestHeavisideAPI_float64):
def setUp(self):
self.x_np = np.random.random((13, 17)).astype("int32")
self.y_np = np.random.random((13, 17)).astype("int32")
self.out_np = np.heaviside(self.x_np, self.y_np)
self.dtype = "int32"
class TestElementwiseOp1(TestElementwiseOp):
def setUp(self):
self.op_type = "elementwise_heaviside"
x = np.random.random(100).astype("float64")
y = np.random.random((13, 100)).astype("float64")
self.python_api = paddle.heaviside
self.prim_op_type = "comp"
self.public_python_api = paddle.heaviside
self.inputs = {'X': x, 'Y': y}
self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])}
class TestElementwiseOp2(TestElementwiseOp):
def setUp(self):
self.op_type = "elementwise_heaviside"
x = np.random.random((13, 100)).astype("float64")
y = np.random.random(100).astype("float64")
self.python_api = paddle.heaviside
self.prim_op_type = "comp"
self.public_python_api = paddle.heaviside
self.inputs = {'X': x, 'Y': y}
self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])}
class TestElementwiseOp3(TestElementwiseOp):
def setUp(self):
self.op_type = "elementwise_heaviside"
x = np.random.uniform(-10, 10, [100]).astype("float64")
y = np.random.uniform(-10, 10, [3, 100]).astype("float64")
self.python_api = paddle.heaviside
self.prim_op_type = "comp"
self.public_python_api = paddle.heaviside
self.inputs = {'X': x, 'Y': y}
self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])}
class TestElementwiseOp4(TestElementwiseOp):
def setUp(self):
self.op_type = "elementwise_heaviside"
x = np.random.uniform(0, 10, []).astype("float64")
y = np.random.uniform(-10, 0, [2, 3, 20]).astype("float64")
self.python_api = paddle.heaviside
self.prim_op_type = "comp"
self.public_python_api = paddle.heaviside
self.inputs = {'X': x, 'Y': y}
self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])}
class TestHeavisideFP16Op(OpTest):
def setUp(self):
self.dtype = np.float16
self.op_type = "elementwise_heaviside"
self.python_api = paddle.heaviside
self.prim_op_type = "comp"
self.public_python_api = paddle.heaviside
self.inputs = {
'X': np.random.uniform(1, 2, [20, 5]).astype("float16"),
'Y': np.random.uniform(1, 2, [20, 5]).astype("float16"),
}
self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])}
def test_check_output(self):
self.check_output(
check_pir=True, check_prim_pir=True, check_symbol_infer=False
)
def test_check_grad(self):
self.check_grad(
['X', 'Y'],
'Out',
user_defined_grads=Heaviside_grad(
self.inputs['X'], self.inputs['Y'], 1 / self.inputs['X'].size
),
check_pir=True,
check_prim_pir=True,
)
@unittest.skipIf(
not (core.is_compiled_with_cuda() or is_custom_device())
or not core.is_bfloat16_supported(get_device_place()),
"core is not compiled with CUDA or not support bfloat16",
)
class TestHeavisideBF16Op(OpTest):
def setUp(self):
self.dtype = np.uint16
self.np_dtype = np.float32
self.op_type = "elementwise_heaviside"
self.python_api = paddle.heaviside
self.prim_op_type = "comp"
self.public_python_api = paddle.heaviside
self.inputs = {
'X': np.random.uniform(1, 2, [20, 5]).astype(self.np_dtype),
'Y': np.random.uniform(1, 2, [20, 5]).astype(self.np_dtype),
}
self.outputs = {'Out': np.heaviside(self.inputs['X'], self.inputs['Y'])}
self.place = get_device_place()
self.inputs['X'] = convert_float_to_uint16(self.inputs['X'])
self.inputs['Y'] = convert_float_to_uint16(self.inputs['Y'])
self.outputs['Out'] = convert_float_to_uint16(self.outputs['Out'])
def test_check_output(self):
self.check_output_with_place(
self.place,
check_pir=True,
check_prim_pir=True,
check_symbol_infer=False,
)
def test_check_grad(self):
self.check_grad_with_place(
self.place,
['X', 'Y'],
'Out',
user_defined_grads=Heaviside_grad(
self.inputs['X'],
self.inputs['Y'],
1 / self.inputs['X'].size,
self.np_dtype,
True,
),
check_pir=True,
check_prim_pir=True,
)
class TestHeavisideError(unittest.TestCase):
def test_input(self):
paddle.disable_static()
def test_input_x():
paddle.heaviside(1, paddle.randn([100]))
self.assertRaises(ValueError, test_input_x)
def test_input_y():
paddle.heaviside(paddle.randn([100]), 1)
self.assertRaises(ValueError, test_input_y)
def test_input_xy():
paddle.heaviside(
paddle.randn([100], 'float32'), paddle.randn([100], 'float64')
)
self.assertRaises(ValueError, test_input_xy)
@unittest.skipIf(
not (core.is_compiled_with_cuda() or is_custom_device()),
"core is not compiled with CUDA",
)
class TestElementwiseHeavisideOp_Stride(OpTest):
no_need_check_grad = True
def setUp(self):
self.op_type = "elementwise_heaviside"
self.python_api = paddle.heaviside
self.public_python_api = paddle.heaviside
self.transpose_api = paddle.transpose
self.as_stride_api = paddle.as_strided
self.init_dtype()
self.init_input_output()
self.inputs_stride = {
'X': OpTest.np_dtype_to_base_dtype(self.x),
'Y': OpTest.np_dtype_to_base_dtype(self.y_trans),
}
self.inputs = {
'X': OpTest.np_dtype_to_base_dtype(self.x),
'Y': OpTest.np_dtype_to_base_dtype(self.y),
}
self.outputs = {'Out': self.out}
def init_dtype(self):
self.dtype = np.float64
self.val_dtype = np.float64
def test_check_output(self):
place = get_device_place()
self.check_strided_forward = True
self.check_output(
place,
)
def init_input_output(self):
self.strided_input_type = "transpose"
self.x = np.random.uniform(0.1, 1, [13, 17]).astype(self.dtype)
self.y = np.random.uniform(0.1, 1, [13, 17]).astype(self.dtype)
self.out = np.heaviside(self.x, self.y)
self.perm = [1, 0]
self.y_trans = np.transpose(self.y, self.perm)
def test_check_gradient(self):
pass
class TestElementwiseHeavisideOp_Stride1(TestElementwiseHeavisideOp_Stride):
def init_input_output(self):
self.strided_input_type = "transpose"
self.x = np.random.uniform(0.1, 1, [20, 2, 13, 17]).astype(self.dtype)
self.y = np.random.uniform(0.1, 1, [20, 2, 13, 17]).astype(self.dtype)
self.out = np.heaviside(self.x, self.y)
self.perm = [0, 1, 3, 2]
self.y_trans = np.transpose(self.y, self.perm)
class TestElementwiseHeavisideOp_Stride2(TestElementwiseHeavisideOp_Stride):
def init_input_output(self):
self.strided_input_type = "transpose"
self.x = np.random.uniform(0.1, 1, [20, 2, 13, 17]).astype(self.dtype)
self.y = np.random.uniform(0.1, 1, [20, 2, 13, 17]).astype(self.dtype)
self.out = np.heaviside(self.x, self.y)
self.perm = [0, 2, 1, 3]
self.y_trans = np.transpose(self.y, self.perm)
class TestElementwiseHeavisideOp_Stride3(TestElementwiseHeavisideOp_Stride):
def init_input_output(self):
self.strided_input_type = "transpose"
self.x = np.random.uniform(0.1, 1, [20, 2, 13, 17]).astype(self.dtype)
self.y = np.random.uniform(0.1, 1, [20, 2, 13, 1]).astype(self.dtype)
self.out = np.heaviside(self.x, self.y)
self.perm = [0, 1, 3, 2]
self.y_trans = np.transpose(self.y, self.perm)
class TestElementwiseHeavisideOp_Stride4(TestElementwiseHeavisideOp_Stride):
def init_input_output(self):
self.strided_input_type = "transpose"
self.x = np.random.uniform(0.1, 1, [1, 2, 13, 17]).astype(self.dtype)
self.y = np.random.uniform(0.1, 1, [20, 2, 13, 1]).astype(self.dtype)
self.out = np.heaviside(self.x, self.y)
self.perm = [1, 0, 2, 3]
self.y_trans = np.transpose(self.y, self.perm)
class TestElementwiseHeavisideOp_Stride5(TestElementwiseHeavisideOp_Stride):
def init_input_output(self):
self.strided_input_type = "as_stride"
self.x = np.random.uniform(0.1, 1, [23, 10, 1, 17]).astype(self.dtype)
self.y = np.random.uniform(0.1, 1, [23, 2, 13, 20]).astype(self.dtype)
self.y_trans = self.y
self.y = self.y[:, 0:1, :, 0:1]
self.out = np.heaviside(self.x, self.y)
self.shape_param = [23, 1, 13, 1]
self.stride_param = [520, 260, 20, 1]
class TestElementwiseHeavisideOp_Stride_ZeroDim1(
TestElementwiseHeavisideOp_Stride
):
def init_input_output(self):
self.strided_input_type = "transpose"
self.x = np.random.uniform(0.1, 1, []).astype(self.dtype)
self.y = np.random.uniform(0.1, 1, [13, 17]).astype(self.dtype)
self.out = np.heaviside(self.x, self.y)
self.perm = [1, 0]
self.y_trans = np.transpose(self.y, self.perm)
class TestElementwiseHeavisideOp_Stride_ZeroSize1(
TestElementwiseHeavisideOp_Stride
):
def init_data(self):
self.strided_input_type = "transpose"
self.x = np.random.rand(1, 0, 2).astype('float32')
self.y = np.random.rand(3, 0, 1).astype('float32')
self.out = np.heaviside(self.x, self.y)
self.perm = [2, 1, 0]
self.y_trans = np.transpose(self.y, self.perm)
@unittest.skipIf(
not (core.is_compiled_with_cuda() or is_custom_device()),
"core is not compiled with CUDA",
)
class TestHeavisideZeroSizeTensor(unittest.TestCase):
"""Regression test for 0-size tensor in paddle.heaviside forward and backward.
When input has a dimension of 0, the broadcast backward kernels
(ElemwiseGradBroadcast1CUDA / ElemwiseGradBroadcast2CUDA) were incorrectly
launched with block_size=0 or grid_size=0, causing CUDA error(9)
(cudaErrorInvalidConfiguration).
Fix: add early-return guards in ElemwiseGradBroadcast1CUDA (h==0 || w==0)
and ElemwiseGradBroadcast2CUDA (pre==0 || n==0 || post==0).
"""
def setUp(self):
self.place = get_device_place()
paddle.disable_static(place=self.place)
def _check_forward_backward(self, x_shape, y_shape, dtype='float32'):
"""Run forward + backward and assert output shape and no CUDA error."""
x = paddle.zeros(x_shape, dtype=dtype)
y = paddle.ones(y_shape, dtype=dtype)
x.stop_gradient = False
y.stop_gradient = False
out = paddle.heaviside(x, y)
expected_shape = list(
np.broadcast_shapes(tuple(x_shape), tuple(y_shape))
)
self.assertEqual(list(out.shape), expected_shape)
out_grad = paddle.ones_like(out)
grads = paddle.grad(
[out],
[x, y],
grad_outputs=[out_grad],
allow_unused=True,
)
self.assertEqual(list(grads[0].shape), x_shape)
self.assertEqual(list(grads[1].shape), y_shape)
# Verify no sticky CUDA error was left by any kernel launch
core.eager._for_test_check_cuda_error()
return out
def _check_forward_only(self, x_shape, y_shape, dtype='int32'):
"""Run forward-only for non-float dtypes and assert shape + no CUDA error."""
x = paddle.zeros(x_shape, dtype=dtype)
y = paddle.ones(y_shape, dtype=dtype)
out = paddle.heaviside(x, y)
expected_shape = list(
np.broadcast_shapes(tuple(x_shape), tuple(y_shape))
)
self.assertEqual(list(out.shape), expected_shape)
core.eager._for_test_check_cuda_error()
return out
# ---------------------------------------------------------------
# Same-shape 0-size (no broadcast) — ElemwiseGradComputeNoBroadcast
# ---------------------------------------------------------------
def test_same_shape_zero_leading_dim_float32(self):
"""[0, 2048] x [0, 2048] same shape, no broadcast."""
self._check_forward_backward([0, 2048], [0, 2048])
def test_same_shape_zero_leading_dim_float64(self):
"""[0, 17] x [0, 17]."""
self._check_forward_backward([0, 17], [0, 17], 'float64')
def test_same_shape_zero_trailing_dim_float64(self):
"""[13, 0] x [13, 0]."""
self._check_forward_backward([13, 0], [13, 0], 'float64')
def test_same_shape_zero_trailing_dim_int32(self):
"""[13, 0] x [13, 0] int32, forward only."""
self._check_forward_only([13, 0], [13, 0], 'int32')
def test_same_shape_zero_trailing_dim_int64(self):
"""[13, 0] x [13, 0] int64, forward only."""
self._check_forward_only([13, 0], [13, 0], 'int64')
def test_same_shape_zero_leading_dim_int32(self):
"""[0, 17] x [0, 17] int32, forward only."""
self._check_forward_only([0, 17], [0, 17], 'int32')
def test_same_shape_zero_leading_dim_int64(self):
"""[0, 17] x [0, 17] int64, forward only."""
self._check_forward_only([0, 17], [0, 17], 'int64')
# ---------------------------------------------------------------
# ElemwiseGradBroadcast1CUDA — h=0 (block_size would be 0)
# ---------------------------------------------------------------
def test_broadcast1_zero_trailing_dim_scalar(self):
"""[300, 0] x [1] → Broadcast1CUDA(h=pre=0, w=n=1), block_size=0."""
self._check_forward_backward([300, 0], [1])
def test_broadcast1_zero_leading_dim_scalar(self):
"""[0, 2048] x [1] → Broadcast1CUDA(h=pre=0, w=n=1), block_size=0."""
self._check_forward_backward([0, 2048], [1])
def test_broadcast1_zero_leading_dim_last_dim(self):
"""[0, 2048] x [2048] → Broadcast1CUDA(h=pre=0, w=n=2048), block_size=0."""
self._check_forward_backward([0, 2048], [2048])
def test_broadcast1_scalar_zero_trailing_dim(self):
"""[1] x [300, 0] symmetric of test_broadcast1_zero_trailing_dim_scalar."""
self._check_forward_backward([1], [300, 0])
def test_broadcast1_scalar_zero_leading_dim(self):
"""[1] x [0, 2048] symmetric of test_broadcast1_zero_leading_dim_scalar."""
self._check_forward_backward([1], [0, 2048])
def test_broadcast1_last_dim_zero_leading_dim(self):
"""[2048] x [0, 2048] symmetric of test_broadcast1_zero_leading_dim_last_dim."""
self._check_forward_backward([2048], [0, 2048])
# ---------------------------------------------------------------
# ElemwiseGradBroadcast1CUDA — w=0 (grid_size would be 0)
# ---------------------------------------------------------------
def test_broadcast1_zero_mid_dim_w_zero(self):
"""[2, 0, 3] x [0, 3] → Broadcast1CUDA(h=2, w=0), grid_size=0."""
self._check_forward_backward([2, 0, 3], [0, 3])
# ---------------------------------------------------------------
# ElemwiseGradBroadcast2CUDA — post=0 (block_size would be 0)
# ---------------------------------------------------------------
def test_broadcast2_zero_post_dim(self):
"""[2, 3, 0] x [3, 1] → Broadcast2CUDA(pre=2, n=3, post=0), block_size=0."""
self._check_forward_backward([2, 3, 0], [3, 1])
if __name__ == '__main__':
unittest.main()