wehub-resource-sync
295 lines
8.7 KiB
Python
295 lines
8.7 KiB
Python
# Copyright (c) 2023 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,
|
|
convert_uint16_to_float,
|
|
get_device_place,
|
|
get_numeric_gradient,
|
|
is_custom_device,
|
|
)
|
|
from testsuite import create_op
|
|
|
|
import paddle
|
|
from paddle.base import core
|
|
|
|
|
|
def fractional_rational_u(u, alpha, input, output, pool_size=0):
|
|
if pool_size > 0:
|
|
return u
|
|
|
|
base = input // output
|
|
|
|
u_max1 = (base + 2) / alpha - 1
|
|
u_max2 = (input + 1 - base) / alpha - (output - 1)
|
|
max_u = min(u_max1, u_max2)
|
|
|
|
return u * max_u
|
|
|
|
|
|
def fractional_start_index(idx, alpha, u, pool_size=0):
|
|
return int((idx + u) * alpha) - int(u * alpha)
|
|
|
|
|
|
def fractional_end_index(idx, alpha, u, pool_size=0):
|
|
if pool_size > 0:
|
|
return int((idx + u) * alpha) - int(u * alpha) + pool_size
|
|
return int((idx + 1 + u) * alpha) - int(u * alpha)
|
|
|
|
|
|
def fractional_max_pool2D_forward_naive(
|
|
x,
|
|
output_size,
|
|
kernel_size=[0, 0],
|
|
random_u=None,
|
|
return_mask=True,
|
|
):
|
|
N, C, H, W = x.shape
|
|
H_out, W_out = output_size
|
|
pool_height, pool_width = kernel_size
|
|
|
|
u = random_u
|
|
|
|
alpha_height = (H - pool_height) / (H_out - (1 if pool_height > 0 else 0))
|
|
alpha_width = (W - pool_width) / (W_out - (1 if pool_width > 0 else 0))
|
|
|
|
u_height = fractional_rational_u(u, alpha_height, H, H_out, pool_height)
|
|
u_width = fractional_rational_u(u, alpha_width, W, W_out, pool_width)
|
|
|
|
out = np.zeros((N, C, H_out, W_out))
|
|
mask = np.zeros((N, C, H_out, W_out))
|
|
for i in range(H_out):
|
|
h_start = fractional_start_index(i, alpha_height, u_height, pool_height)
|
|
h_end = fractional_end_index(i, alpha_height, u_height, pool_height)
|
|
h_start = max(h_start, 0)
|
|
h_end = min(h_end, H)
|
|
|
|
for j in range(W_out):
|
|
w_start = fractional_start_index(
|
|
j, alpha_width, u_width, pool_width
|
|
)
|
|
w_end = fractional_end_index(j, alpha_width, u_width, pool_width)
|
|
w_start = max(w_start, 0)
|
|
w_end = min(w_end, W)
|
|
|
|
x_masked = x[:, :, h_start:h_end, w_start:w_end]
|
|
|
|
out[:, :, i, j] = np.max(x_masked, axis=(2, 3))
|
|
|
|
for n in range(N):
|
|
for c in range(C):
|
|
arr = x_masked[n, c, :, :]
|
|
index = np.where(arr == np.max(arr))
|
|
sub_row = index[0][0]
|
|
sub_col = index[1][0]
|
|
index = (h_start + sub_row) * W + w_start + sub_col
|
|
mask[n, c, i, j] = index
|
|
|
|
return out, mask
|
|
|
|
|
|
# ----------------fractional_max_pool2d----------------
|
|
def fractional_max_pool2d_wrapper(
|
|
x,
|
|
output_size=None,
|
|
kernel_size=[0, 0],
|
|
random_u=None,
|
|
return_mask=True,
|
|
):
|
|
return paddle._C_ops.fractional_max_pool2d(
|
|
x,
|
|
output_size,
|
|
kernel_size,
|
|
random_u,
|
|
return_mask,
|
|
)
|
|
|
|
|
|
class TestMaxPoolWithIndex_Op(OpTest):
|
|
def setUp(self):
|
|
self.op_type = "fractional_max_pool2d"
|
|
self.python_api = fractional_max_pool2d_wrapper
|
|
self.pool_forward_naive = fractional_max_pool2D_forward_naive
|
|
|
|
self.init_test_case()
|
|
self.init_fractional()
|
|
self.init_dtype()
|
|
|
|
if self.is_bfloat16_op():
|
|
np.random.seed(2023)
|
|
input = np.random.random(self.shape).astype(np.float32)
|
|
input = convert_uint16_to_float(
|
|
convert_float_to_uint16(np.round(input * 100.0, 2))
|
|
)
|
|
|
|
else:
|
|
np.random.seed(2023)
|
|
input = np.random.random(self.shape).astype(self.dtype)
|
|
input = np.round(input * 100.0, 2)
|
|
|
|
output, mask = self.pool_forward_naive(
|
|
input,
|
|
self.output_size,
|
|
self.kernel_size,
|
|
self.random_u,
|
|
self.return_mask,
|
|
)
|
|
mask = mask.astype("int32")
|
|
if self.is_bfloat16_op():
|
|
output = output.astype(np.float32)
|
|
else:
|
|
output = output.astype(self.dtype)
|
|
|
|
self.attrs = {
|
|
'output_size': self.output_size,
|
|
'kernel_size': self.kernel_size,
|
|
'random_u': self.random_u,
|
|
'return_mask': self.return_mask,
|
|
}
|
|
|
|
if self.is_bfloat16_op():
|
|
self.inputs = {'x': convert_float_to_uint16(input)}
|
|
self.outputs = {
|
|
'out': convert_float_to_uint16(output),
|
|
'mask': mask,
|
|
}
|
|
self.inputs_fp32 = {'x': input}
|
|
|
|
else:
|
|
self.inputs = {'x': input}
|
|
self.outputs = {'out': output, 'mask': mask}
|
|
|
|
def init_dtype(self):
|
|
self.dtype = np.float64
|
|
|
|
def test_check_output(self):
|
|
self.check_output()
|
|
|
|
def test_check_grad(self):
|
|
self.check_grad({'x'}, ['out'])
|
|
|
|
def init_test_case(self):
|
|
self.shape = [2, 3, 7, 7]
|
|
self.output_size = [3, 3]
|
|
self.kernel_size = [0, 0]
|
|
self.return_mask = True
|
|
|
|
def init_fractional(self):
|
|
self.random_u = 0.3
|
|
|
|
|
|
class TestCase1(TestMaxPoolWithIndex_Op):
|
|
def init_test_case(self):
|
|
self.shape = [3, 5, 9, 9]
|
|
self.output_size = [5, 5]
|
|
self.kernel_size = [0, 0]
|
|
self.return_mask = False
|
|
|
|
|
|
class TestCase2(TestCase1):
|
|
def init_fractional(self):
|
|
self.random_u = 0.5
|
|
|
|
|
|
class TestCase3(TestMaxPoolWithIndex_Op):
|
|
def init_test_case(self):
|
|
self.shape = [3, 5, 9, 9]
|
|
self.output_size = [7, 7]
|
|
self.kernel_size = [2, 2]
|
|
self.return_mask = True
|
|
|
|
|
|
# ----------------fractional_max_pool2d_fp16----------------
|
|
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 TestMaxPool2dFP16(parent):
|
|
def init_dtype(self):
|
|
self.dtype = np.float16
|
|
|
|
def test_check_output(self):
|
|
if core.is_compiled_with_cuda() or is_custom_device():
|
|
place = get_device_place()
|
|
if core.is_float16_supported(place):
|
|
self.check_output_with_place(place)
|
|
|
|
def test_check_grad(self):
|
|
place = get_device_place()
|
|
if core.is_float16_supported(place):
|
|
self.check_grad_with_place(place, {'x'}, ['out'])
|
|
|
|
cls_name = "{}_{}".format(parent.__name__, "FP16OP")
|
|
TestMaxPool2dFP16.__name__ = cls_name
|
|
globals()[cls_name] = TestMaxPool2dFP16
|
|
|
|
|
|
create_test_fp16_class(TestMaxPoolWithIndex_Op)
|
|
create_test_fp16_class(TestCase1)
|
|
create_test_fp16_class(TestCase2)
|
|
create_test_fp16_class(TestCase3)
|
|
|
|
|
|
# ----------------fractional_max_pool2d_bf16----------------
|
|
def create_test_bf16_class(parent):
|
|
@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 and do not support bfloat16",
|
|
)
|
|
class TestMaxPool2dBF16(parent):
|
|
def init_dtype(self):
|
|
self.dtype = np.uint16
|
|
|
|
def get_numeric_grad(self, place, check_name):
|
|
scope = core.Scope()
|
|
self._check_grad_helper()
|
|
op = create_op(
|
|
scope, self.op_type, self.inputs, self.outputs, self.attrs
|
|
)
|
|
return get_numeric_gradient(
|
|
place, scope, op, self.inputs_fp32, check_name, ['out']
|
|
)
|
|
|
|
def test_check_output(self):
|
|
place = get_device_place()
|
|
if core.is_bfloat16_supported(place):
|
|
self.check_output_with_place(place)
|
|
|
|
def test_check_grad(self):
|
|
place = get_device_place()
|
|
numeric_grads = self.get_numeric_grad(place, 'x')
|
|
if core.is_bfloat16_supported(place):
|
|
self.check_grad_with_place(
|
|
place, {'x'}, ['out'], user_defined_grads=[numeric_grads]
|
|
)
|
|
|
|
cls_name = "{}_{}".format(parent.__name__, "BF16OP")
|
|
TestMaxPool2dBF16.__name__ = cls_name
|
|
globals()[cls_name] = TestMaxPool2dBF16
|
|
|
|
|
|
create_test_bf16_class(TestMaxPoolWithIndex_Op)
|
|
create_test_bf16_class(TestCase1)
|
|
create_test_bf16_class(TestCase2)
|
|
create_test_bf16_class(TestCase3)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|