wehub-resource-sync
2455 lines
91 KiB
Python
2455 lines
91 KiB
Python
# 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 math
|
|
import unittest
|
|
|
|
import numpy as np
|
|
from op_test import get_device_place, is_custom_device
|
|
from scipy import special
|
|
from utils import dygraph_guard, static_guard
|
|
|
|
import paddle
|
|
from paddle.base import framework
|
|
from paddle.base.core import VarDesc
|
|
from paddle.regularizer import L2Decay
|
|
|
|
DELTA = 0.00001
|
|
|
|
|
|
def check_cast_op(op):
|
|
return (
|
|
op.type == 'cast'
|
|
and op.attr('in_dtype') == VarDesc.VarType.FP32
|
|
and op.attr('out_dtype') in [VarDesc.VarType.FP16, VarDesc.VarType.BF16]
|
|
)
|
|
|
|
|
|
def check_cast_op_pir(op):
|
|
return (
|
|
op.name() == 'pd_op.cast'
|
|
and op.attrs()['dtype']
|
|
in (
|
|
paddle.base.libpaddle.DataType.FLOAT16,
|
|
paddle.base.libpaddle.DataType.BFLOAT16,
|
|
)
|
|
and op.operand_source(0).dtype == paddle.base.libpaddle.DataType.FLOAT32
|
|
)
|
|
|
|
|
|
def output_hist(out):
|
|
hist, _ = np.histogram(out, range=(-1, 1))
|
|
hist = hist.astype("float32")
|
|
hist /= float(out.size)
|
|
prob = 0.1 * np.ones(10)
|
|
return hist, prob
|
|
|
|
|
|
class TestConstantInitializer(unittest.TestCase):
|
|
def test_calculate_gain(self):
|
|
self.assertEqual(paddle.nn.initializer.calculate_gain('sigmoid'), 1)
|
|
self.assertEqual(paddle.nn.initializer.calculate_gain('linear'), 1)
|
|
self.assertEqual(paddle.nn.initializer.calculate_gain('conv2d'), 1)
|
|
self.assertEqual(paddle.nn.initializer.calculate_gain('tanh'), 5.0 / 3)
|
|
self.assertEqual(
|
|
paddle.nn.initializer.calculate_gain('relu'), math.sqrt(2.0)
|
|
)
|
|
self.assertEqual(
|
|
paddle.nn.initializer.calculate_gain('leaky_relu', 1), 1
|
|
)
|
|
self.assertEqual(paddle.nn.initializer.calculate_gain('selu'), 3.0 / 4)
|
|
|
|
def test_constant_initializer_default_value(self, dtype="float32"):
|
|
"""Test the constant initializer with default value"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.Constant(),
|
|
)
|
|
num_ops = 1
|
|
self.assertEqual(len(block.ops), num_ops)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'fill_constant')
|
|
self.assertAlmostEqual(init_op.attr('value'), 0.0, delta=DELTA)
|
|
return block
|
|
|
|
def test_constant_initializer(self, dtype="float32"):
|
|
"""Test constant initializer with supplied value"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.Constant(2.3),
|
|
)
|
|
num_ops = 1
|
|
self.assertEqual(len(block.ops), num_ops)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'fill_constant')
|
|
self.assertAlmostEqual(init_op.attr('value'), 2.3, delta=DELTA)
|
|
return block
|
|
|
|
def test_constant_initializer_fp16(self):
|
|
"""Test constant initializer with float16"""
|
|
self.test_constant_initializer_default_value("float16")
|
|
self.test_constant_initializer("float16")
|
|
|
|
def test_constant_initializer_bf16(self):
|
|
"""Test constant initializer with bfloat16
|
|
No cast operator has been added here
|
|
"""
|
|
self.test_constant_initializer_default_value("uint16")
|
|
self.test_constant_initializer("uint16")
|
|
|
|
|
|
class TestUniformInitializer(unittest.TestCase):
|
|
def test_uniform_initializer_default_value(self, dtype="float32"):
|
|
"""Test the uniform initializer with default value"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.Uniform(),
|
|
)
|
|
num_ops = 2 if dtype == "float16" else 1
|
|
self.assertEqual(len(block.ops), num_ops)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'uniform_random')
|
|
self.assertAlmostEqual(init_op.attr('min'), -1.0, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('max'), 1.0, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 0)
|
|
return block
|
|
|
|
def test_uniform_initializer_random_seed(self):
|
|
"""Test the uniform initializer with manually setting seed"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
program.random_seed = 123
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param1",
|
|
initializer=paddle.nn.initializer.Uniform(),
|
|
)
|
|
block.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param2",
|
|
initializer=paddle.nn.initializer.UniformInitializer(
|
|
seed=456
|
|
),
|
|
)
|
|
init_op = block.ops[1]
|
|
self.assertEqual(init_op.attr("seed"), 456)
|
|
init_op1 = block.ops[0]
|
|
self.assertEqual(init_op1.attr("seed"), 123)
|
|
|
|
def test_uniform_initializer(self, dtype="float32"):
|
|
"""Test uniform initializer with supplied attributes"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.UniformInitializer(
|
|
-4.2, 3.1, 123
|
|
),
|
|
)
|
|
num_ops = 2 if dtype == "float16" else 1
|
|
self.assertEqual(len(block.ops), num_ops)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'uniform_random')
|
|
self.assertAlmostEqual(init_op.attr('min'), -4.2, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('max'), 3.1, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 123)
|
|
return block
|
|
|
|
def test_uniform_initializer_two_op(self, dtype="float32"):
|
|
"""Test uniform initializer with supplied attributes"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for i in range(2):
|
|
block.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.UniformInitializer(
|
|
-4.2, float(i), 123
|
|
),
|
|
)
|
|
num_ops = 2 if dtype == "float16" else 1
|
|
self.assertEqual(len(block.ops), num_ops)
|
|
init_op0 = block.ops[0]
|
|
self.assertEqual(init_op0.type, 'uniform_random')
|
|
self.assertAlmostEqual(init_op0.attr('min'), -4.2, delta=DELTA)
|
|
self.assertAlmostEqual(init_op0.attr('max'), 0.0, delta=DELTA)
|
|
self.assertEqual(init_op0.attr('seed'), 123)
|
|
return block
|
|
|
|
def test_uniform_initializer_fp16(self):
|
|
"""Test uniform initializer with float16"""
|
|
block = self.test_uniform_initializer_default_value("float16")
|
|
self.assertTrue(check_cast_op(block.ops[1]))
|
|
block = self.test_uniform_initializer(dtype="float16")
|
|
self.assertTrue(check_cast_op(block.ops[1]))
|
|
block = self.test_uniform_initializer_two_op("float16")
|
|
self.assertTrue(check_cast_op(block.ops[1]))
|
|
|
|
def test_uniform_initializer_bf16(self):
|
|
"""Test uniform initializer with bfloat16
|
|
No cast operator has been added here
|
|
"""
|
|
block = self.test_uniform_initializer_default_value("uint16")
|
|
block = self.test_uniform_initializer(dtype="uint16")
|
|
block = self.test_uniform_initializer_two_op("uint16")
|
|
|
|
|
|
class TestUniformInitializerPir(unittest.TestCase):
|
|
def setUp(self):
|
|
self.init_op_name = 'pd_op.uniform'
|
|
self.set_parameter_op_name = 'builtin.set_parameter'
|
|
|
|
def get_operand_definition_op_attrs(self, cur_op, operand_name, attr_name):
|
|
input_names = cur_op.get_input_names()
|
|
self.assertIn(operand_name, input_names)
|
|
attr = (
|
|
cur_op.operand(input_names.index(operand_name))
|
|
.source()
|
|
.get_defining_op()
|
|
.attrs()[attr_name]
|
|
)
|
|
return attr
|
|
|
|
def test_uniform_initializer_default_value(self, dtype="float32"):
|
|
"""Test the uniform initializer with default value"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.Uniform(),
|
|
)
|
|
block = startup.global_block()
|
|
for op in block.ops:
|
|
# get init op
|
|
if self.init_op_name == op.name():
|
|
min = self.get_operand_definition_op_attrs(
|
|
op, "min", "value"
|
|
)
|
|
max = self.get_operand_definition_op_attrs(
|
|
op, "max", "value"
|
|
)
|
|
self.assertAlmostEqual(min, -1.0, delta=DELTA)
|
|
self.assertAlmostEqual(max, 1.0, delta=DELTA)
|
|
self.assertEqual(op.attrs()['seed'], 0)
|
|
|
|
def test_uniform_initializer_random_seed(self):
|
|
"""Test the uniform initializer with manually setting seed"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
startup.random_seed = 123
|
|
with paddle.static.program_guard(main, startup):
|
|
param1 = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param1",
|
|
initializer=paddle.nn.initializer.Uniform(),
|
|
)
|
|
|
|
param2 = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param2",
|
|
initializer=paddle.nn.initializer.UniformInitializer(
|
|
seed=456
|
|
),
|
|
)
|
|
|
|
block = startup.global_block()
|
|
|
|
checked_parameter_names = []
|
|
for op in block.ops:
|
|
if self.set_parameter_op_name != op.name():
|
|
continue
|
|
|
|
parameter_name = op.attrs()["parameter_name"]
|
|
if parameter_name == "param1":
|
|
# get "param1"
|
|
checked_parameter_names.append(parameter_name)
|
|
seed = (
|
|
op.operand(0)
|
|
.source()
|
|
.get_defining_op()
|
|
.attrs()['seed']
|
|
)
|
|
self.assertEqual(seed, 123)
|
|
elif parameter_name == "param2":
|
|
# get "param2"
|
|
checked_parameter_names.append(parameter_name)
|
|
seed = (
|
|
op.operand(0)
|
|
.source()
|
|
.get_defining_op()
|
|
.attrs()['seed']
|
|
)
|
|
self.assertEqual(seed, 456)
|
|
|
|
self.assertIn("param1", checked_parameter_names)
|
|
self.assertIn("param2", checked_parameter_names)
|
|
|
|
def test_uniform_initializer(self, dtype="float32"):
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
initializer = paddle.nn.initializer.UniformInitializer(
|
|
low=-0.5,
|
|
high=0.5,
|
|
seed=10,
|
|
diag_num=16,
|
|
diag_step=16,
|
|
diag_val=1.0,
|
|
)
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=initializer,
|
|
)
|
|
block = startup.global_block()
|
|
for op in block.ops:
|
|
# get init op
|
|
if self.init_op_name == op.name():
|
|
self.assertEqual(op.attrs()["seed"], 10)
|
|
|
|
input_names = op.get_input_names()
|
|
self.assertIn('shape', input_names)
|
|
self.assertIn('min', input_names)
|
|
self.assertIn('max', input_names)
|
|
shape = self.get_operand_definition_op_attrs(
|
|
op, "shape", "value"
|
|
)
|
|
min = self.get_operand_definition_op_attrs(
|
|
op, "min", "value"
|
|
)
|
|
max = self.get_operand_definition_op_attrs(
|
|
op, "max", "value"
|
|
)
|
|
self.assertEqual(shape, [5, 10])
|
|
self.assertAlmostEqual(min, -0.5, DELTA)
|
|
self.assertAlmostEqual(max, 0.5, DELTA)
|
|
|
|
def test_uniform_initializer_fp16(self):
|
|
"""Test uniform initializer with float16"""
|
|
self.test_uniform_initializer_default_value(dtype="float16")
|
|
self.test_uniform_initializer(dtype="float16")
|
|
|
|
def test_uniform_initializer_bf16(self):
|
|
"""Test uniform initializer with float16"""
|
|
self.test_uniform_initializer_default_value(dtype="uint16")
|
|
self.test_uniform_initializer(dtype="uint16")
|
|
|
|
|
|
class TestNormalInitializer(unittest.TestCase):
|
|
def test_normal_initializer_default_value(self):
|
|
"""Test the normal initializer with default value"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.Normal(),
|
|
)
|
|
self.assertEqual(len(block.ops), 1)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'gaussian_random')
|
|
self.assertAlmostEqual(init_op.attr('mean'), 0.0, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('std'), 1.0, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 0)
|
|
|
|
def test_normal_initializer(self, dtype="float32"):
|
|
"""Test normal initializer with supplied attributes"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.NormalInitializer(
|
|
2.3, 1.9, 123
|
|
),
|
|
)
|
|
num_ops = 1
|
|
self.assertEqual(len(block.ops), num_ops)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'gaussian_random')
|
|
self.assertAlmostEqual(init_op.attr('mean'), 2.3, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('std'), 1.9, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 123)
|
|
return block
|
|
|
|
def test_normal_initializer_complex(self, dtype="complex64"):
|
|
"""Test normal initializer with complex dtype"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.NormalInitializer(
|
|
2.2 + 2.2j, 1.9, 123
|
|
),
|
|
)
|
|
num_ops = 1
|
|
self.assertEqual(len(block.ops), num_ops)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'gaussian_random')
|
|
self.assertAlmostEqual(init_op.attr('mean'), 2.2, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('std'), 1.9, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 123)
|
|
return block
|
|
|
|
def test_normal_initializer_fp16(self):
|
|
"""Test normal initializer with float16"""
|
|
self.test_normal_initializer("float16")
|
|
|
|
def test_normal_initializer_bf16(self):
|
|
"""Test normal initializer with bfloat16"""
|
|
self.test_normal_initializer("uint16")
|
|
|
|
def test_normal_initializer_complex64(self):
|
|
"""Test normal initializer with complex64"""
|
|
self.test_normal_initializer_complex("complex64")
|
|
|
|
def test_normal_initializer_complex128(self):
|
|
"""Test normal initializer with complex128"""
|
|
self.test_normal_initializer_complex("complex128")
|
|
|
|
|
|
class TestXavierInitializer(unittest.TestCase):
|
|
def test_uniform_xavier_initializer(self):
|
|
"""Test Xavier initializer with uniform distribution on
|
|
for matrix multiply.
|
|
"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
param = block.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierUniform(),
|
|
)
|
|
self.assertEqual(len(block.ops), 1)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'uniform_random')
|
|
limit = np.sqrt(6.0 / (param.shape[0] + param.shape[1]))
|
|
self.assertAlmostEqual(init_op.attr('min'), -limit, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('max'), limit, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 0)
|
|
|
|
def test_uniform_xavier_initializer_conv(self):
|
|
"""Test Xavier initializer with uniform distribution on
|
|
for convolutions.
|
|
"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
param = block.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10, 15, 20],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierUniform(),
|
|
)
|
|
self.assertEqual(len(block.ops), 1)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'uniform_random')
|
|
receptive_field_size = float(15 * 20)
|
|
limit = np.sqrt(
|
|
6.0 / ((param.shape[0] + param.shape[1]) * receptive_field_size)
|
|
)
|
|
self.assertAlmostEqual(init_op.attr('min'), -limit, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('max'), limit, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 0)
|
|
|
|
def test_normal_xavier_initializer(self):
|
|
"""Test Xavier initializer with normal distribution on
|
|
for matrix multiply.
|
|
"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
param = block.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierNormal(),
|
|
)
|
|
self.assertEqual(len(block.ops), 1)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'gaussian_random')
|
|
std = np.sqrt(2.0 / (param.shape[0] + param.shape[1]))
|
|
self.assertAlmostEqual(init_op.attr('mean'), 0.0, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('std'), std, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 0)
|
|
|
|
def test_normal_xavier_initializer_conv(self):
|
|
"""Test Xavier initializer with normal distribution on
|
|
for convolutions.
|
|
"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
param = block.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10, 15, 20],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierNormal(),
|
|
)
|
|
self.assertEqual(len(block.ops), 1)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'gaussian_random')
|
|
receptive_field_size = float(15 * 20)
|
|
std = np.sqrt(
|
|
2.0 / ((param.shape[0] + param.shape[1]) * receptive_field_size)
|
|
)
|
|
self.assertAlmostEqual(init_op.attr('mean'), 0.0, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('std'), std, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 0)
|
|
|
|
def test_xavier_initializer_supplied_arguments(
|
|
self, dtype="float32", uniform=True
|
|
):
|
|
"""Test the Xavier initializer with supplied arguments"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierInitializer(
|
|
uniform=uniform,
|
|
fan_in=12,
|
|
fan_out=23,
|
|
seed=134,
|
|
gain=0.2,
|
|
),
|
|
)
|
|
num_ops = (
|
|
2
|
|
if (dtype == "float16" or (dtype == "uint16" and not uniform))
|
|
else 1
|
|
)
|
|
self.assertEqual(len(block.ops), num_ops)
|
|
init_op = block.ops[0]
|
|
if uniform:
|
|
self.assertEqual(init_op.type, 'uniform_random')
|
|
limit = 0.2 * np.sqrt(6.0 / (12 + 23))
|
|
self.assertAlmostEqual(init_op.attr('min'), -limit, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('max'), limit, delta=DELTA)
|
|
else:
|
|
self.assertEqual(init_op.type, 'gaussian_random')
|
|
self.assertEqual(init_op.attr('seed'), 134)
|
|
return block
|
|
|
|
def test_xavier_initializer_fp16(self):
|
|
"""Test the Xavier initializer with float16"""
|
|
block = self.test_xavier_initializer_supplied_arguments("float16")
|
|
|
|
def test_xavier_initializer_bf16(self):
|
|
"""Test the Xavier initializer with bfloat16"""
|
|
block_uniform = self.test_xavier_initializer_supplied_arguments(
|
|
"uint16"
|
|
)
|
|
self.assertEqual(len(block_uniform.ops), 1)
|
|
block_gaussian = self.test_xavier_initializer_supplied_arguments(
|
|
"uint16", False
|
|
)
|
|
|
|
|
|
class TestXavierInitializerPir(unittest.TestCase):
|
|
def setUp(self):
|
|
self.init_uniform_op_name = 'pd_op.uniform'
|
|
self.init_normal_op_name = 'pd_op.gaussian'
|
|
self.set_parameter_op_name = 'builtin.set_parameter'
|
|
|
|
def get_operand_definition_op_attrs(self, cur_op, operand_name, attr_name):
|
|
input_names = cur_op.get_input_names()
|
|
self.assertIn(operand_name, input_names)
|
|
attr = (
|
|
cur_op.operand(input_names.index(operand_name))
|
|
.source()
|
|
.get_defining_op()
|
|
.attrs()[attr_name]
|
|
)
|
|
return attr
|
|
|
|
def get_init_ops_by_op_name(self, block, op_name):
|
|
checked_ops = []
|
|
for op in block.ops:
|
|
# get init op
|
|
if op_name == op.name():
|
|
checked_ops.append(op)
|
|
return checked_ops
|
|
|
|
def test_uniform_xavier_initializer(self):
|
|
"""Test Xavier initializer with uniform distribution on
|
|
for matrix multiply.
|
|
"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierUniform(),
|
|
)
|
|
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_uniform_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
limit = np.sqrt(6.0 / (param.shape[0] + param.shape[1]))
|
|
min = self.get_operand_definition_op_attrs(
|
|
init_op, "min", "value"
|
|
)
|
|
max = self.get_operand_definition_op_attrs(
|
|
init_op, "max", "value"
|
|
)
|
|
self.assertAlmostEqual(min, -limit, delta=DELTA)
|
|
self.assertAlmostEqual(max, limit, delta=DELTA)
|
|
self.assertEqual(init_op.attrs()['seed'], 0)
|
|
|
|
def test_uniform_xavier_initializer_zero_size(self):
|
|
"""Test Xavier initializer with uniform distribution on
|
|
for matrix multiply.
|
|
"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[0, 0],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierUniform(),
|
|
)
|
|
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_uniform_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
limit = 0.0
|
|
min = self.get_operand_definition_op_attrs(
|
|
init_op, "min", "value"
|
|
)
|
|
max = self.get_operand_definition_op_attrs(
|
|
init_op, "max", "value"
|
|
)
|
|
self.assertAlmostEqual(min, -limit, delta=DELTA)
|
|
self.assertAlmostEqual(max, limit, delta=DELTA)
|
|
self.assertEqual(init_op.attrs()['seed'], 0)
|
|
|
|
def test_uniform_xavier_initializer_conv(self):
|
|
"""Test Xavier initializer with uniform distribution on
|
|
for convolutions.
|
|
"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10, 15, 20],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierUniform(),
|
|
)
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_uniform_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
receptive_field_size = float(15 * 20)
|
|
limit = np.sqrt(
|
|
6.0
|
|
/ ((param.shape[0] + param.shape[1]) * receptive_field_size)
|
|
)
|
|
min = self.get_operand_definition_op_attrs(
|
|
init_op, "min", "value"
|
|
)
|
|
max = self.get_operand_definition_op_attrs(
|
|
init_op, "max", "value"
|
|
)
|
|
self.assertAlmostEqual(min, -limit, delta=DELTA)
|
|
self.assertAlmostEqual(max, limit, delta=DELTA)
|
|
self.assertEqual(init_op.attrs()['seed'], 0)
|
|
|
|
def test_normal_xavier_initializer(self):
|
|
"""Test Xavier initializer with normal distribution on
|
|
for matrix multiply.
|
|
"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierNormal(),
|
|
)
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_normal_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
std = np.sqrt(2.0 / (param.shape[0] + param.shape[1]))
|
|
self.assertAlmostEqual(
|
|
init_op.attrs()["mean"], 0.0, delta=DELTA
|
|
)
|
|
self.assertAlmostEqual(init_op.attrs()["std"], std, delta=DELTA)
|
|
self.assertEqual(init_op.attrs()['seed'], 0)
|
|
|
|
def test_normal_xavier_initializer_zero_size(self):
|
|
"""Test Xavier initializer with normal distribution on
|
|
for matrix multiply.
|
|
"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[0, 0],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierNormal(),
|
|
)
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_normal_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
std = 0.0
|
|
self.assertAlmostEqual(
|
|
init_op.attrs()["mean"], 0.0, delta=DELTA
|
|
)
|
|
self.assertAlmostEqual(init_op.attrs()["std"], std, delta=DELTA)
|
|
self.assertEqual(init_op.attrs()['seed'], 0)
|
|
|
|
def test_normal_xavier_initializer_conv(self):
|
|
"""Test Xavier initializer with normal distribution on
|
|
for convolutions.
|
|
"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10, 15, 20],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierNormal(),
|
|
)
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_normal_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
receptive_field_size = float(15 * 20)
|
|
std = np.sqrt(
|
|
2.0
|
|
/ ((param.shape[0] + param.shape[1]) * receptive_field_size)
|
|
)
|
|
self.assertAlmostEqual(
|
|
init_op.attrs()['mean'], 0.0, delta=DELTA
|
|
)
|
|
self.assertAlmostEqual(init_op.attrs()['std'], std, delta=DELTA)
|
|
self.assertEqual(init_op.attrs()['seed'], 0)
|
|
|
|
def test_xavier_initializer_supplied_arguments(
|
|
self, dtype="float32", uniform=True
|
|
):
|
|
"""Test the Xavier initializer with supplied arguments"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.XavierInitializer(
|
|
uniform=uniform,
|
|
fan_in=12,
|
|
fan_out=23,
|
|
seed=134,
|
|
gain=0.2,
|
|
),
|
|
)
|
|
block = startup.global_block()
|
|
init_op_name = (
|
|
self.init_uniform_op_name
|
|
if uniform
|
|
else self.init_normal_op_name
|
|
)
|
|
|
|
checked_ops = self.get_init_ops_by_op_name(block, init_op_name)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
if uniform:
|
|
limit = 0.2 * np.sqrt(6.0 / (12 + 23))
|
|
min = self.get_operand_definition_op_attrs(
|
|
init_op, "min", "value"
|
|
)
|
|
max = self.get_operand_definition_op_attrs(
|
|
init_op, "max", "value"
|
|
)
|
|
self.assertAlmostEqual(min, -limit, delta=DELTA)
|
|
self.assertAlmostEqual(max, limit, delta=DELTA)
|
|
|
|
self.assertEqual(init_op.attrs()['seed'], 134)
|
|
|
|
return main, startup
|
|
|
|
@unittest.skipIf(
|
|
not (paddle.is_compiled_with_cuda() or is_custom_device()),
|
|
"core is not compiled with CUDA",
|
|
)
|
|
def test_xavier_initializer_fp16(self):
|
|
"""Test the Xavier initializer with float16"""
|
|
main_1, startup_1 = self.test_xavier_initializer_supplied_arguments(
|
|
"float16"
|
|
)
|
|
with paddle.pir_utils.IrGuard():
|
|
exe = paddle.static.Executor(get_device_place())
|
|
exe.run(startup_1)
|
|
exe.run(main_1)
|
|
|
|
main_2, startup_2 = self.test_xavier_initializer_supplied_arguments(
|
|
"float16", uniform=False
|
|
)
|
|
with paddle.pir_utils.IrGuard():
|
|
exe = paddle.static.Executor(get_device_place())
|
|
exe.run(startup_2)
|
|
exe.run(main_2)
|
|
|
|
@unittest.skipIf(
|
|
not (paddle.base.core.is_compiled_with_cuda() or is_custom_device())
|
|
or not paddle.base.core.is_bfloat16_supported(get_device_place()),
|
|
"core is not compiled with CUDA and do not support bfloat16",
|
|
)
|
|
def test_xavier_initializer_bf16(self):
|
|
"""Test the Xavier initializer with bfloat16"""
|
|
main_1, startup_1 = self.test_xavier_initializer_supplied_arguments(
|
|
"uint16"
|
|
)
|
|
with paddle.pir_utils.IrGuard():
|
|
exe = paddle.static.Executor(get_device_place())
|
|
exe.run(startup_1)
|
|
exe.run(main_1)
|
|
|
|
main_2, startup_2 = self.test_xavier_initializer_supplied_arguments(
|
|
"uint16", False
|
|
)
|
|
with paddle.pir_utils.IrGuard():
|
|
exe = paddle.static.Executor(get_device_place())
|
|
exe.run(startup_2)
|
|
exe.run(main_2)
|
|
|
|
|
|
class TestMSRAInitializer(unittest.TestCase):
|
|
def test_uniform_msra_initializer(self):
|
|
"""Test MSRA initializer with uniform distribution on
|
|
for matrix multiply.
|
|
"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
param = block.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.KaimingUniform(),
|
|
)
|
|
self.assertEqual(len(block.ops), 1)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'uniform_random')
|
|
limit = np.sqrt(6.0 / param.shape[0])
|
|
self.assertAlmostEqual(init_op.attr('min'), -limit, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('max'), limit, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 0)
|
|
|
|
def test_uniform_msra_initializer_conv(self):
|
|
"""Test MSRA initializer with uniform distribution on
|
|
for convolutions.
|
|
"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
param = block.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10, 15, 20],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.KaimingUniform(),
|
|
)
|
|
self.assertEqual(len(block.ops), 1)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'uniform_random')
|
|
receptive_field_size = float(15 * 20)
|
|
limit = np.sqrt(6.0 / (param.shape[1] * receptive_field_size))
|
|
self.assertAlmostEqual(init_op.attr('min'), -limit, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('max'), limit, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 0)
|
|
|
|
def test_normal_msra_initializer(self):
|
|
"""Test MSRA initializer with normal distribution on
|
|
for matrix multiply.
|
|
"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
param = block.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.KaimingNormal(),
|
|
)
|
|
self.assertEqual(len(block.ops), 1)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'gaussian_random')
|
|
std = np.sqrt(2.0 / param.shape[0])
|
|
self.assertAlmostEqual(init_op.attr('mean'), 0.0, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('std'), std, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 0)
|
|
|
|
def test_normal_msra_initializer_conv(self):
|
|
"""Test MSRA initializer with normal distribution on
|
|
for convolutions.
|
|
"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
param = block.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10, 15, 20],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.KaimingNormal(),
|
|
)
|
|
self.assertEqual(len(block.ops), 1)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'gaussian_random')
|
|
receptive_field_size = float(15 * 20)
|
|
std = np.sqrt(2.0 / (param.shape[1] * receptive_field_size))
|
|
self.assertAlmostEqual(init_op.attr('mean'), 0.0, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('std'), std, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 0)
|
|
|
|
def test_msra_initializer_supplied_arguments(self, dtype="float32"):
|
|
"""Test the MSRA initializer with supplied arguments"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.MSRAInitializer(
|
|
fan_in=12, seed=134
|
|
),
|
|
)
|
|
num_ops = 2 if dtype == "float16" else 1
|
|
self.assertEqual(len(block.ops), num_ops)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'uniform_random')
|
|
limit = np.sqrt(6.0 / 12)
|
|
self.assertAlmostEqual(init_op.attr('min'), -limit, delta=DELTA)
|
|
self.assertAlmostEqual(init_op.attr('max'), limit, delta=DELTA)
|
|
self.assertEqual(init_op.attr('seed'), 134)
|
|
return block
|
|
|
|
def test_msra_initializer_fp16(self):
|
|
"""Test the MSRA initializer with float16"""
|
|
block = self.test_msra_initializer_supplied_arguments("float16")
|
|
self.assertTrue(check_cast_op(block.ops[1]))
|
|
|
|
def test_msra_initializer_bf16(self):
|
|
"""Test the MSRA initializer with bfloat16"""
|
|
block = self.test_msra_initializer_supplied_arguments("uint16")
|
|
|
|
|
|
class TestMSRAInitializerPir(unittest.TestCase):
|
|
def setUp(self):
|
|
self.init_uniform_op_name = 'pd_op.uniform'
|
|
self.init_normal_op_name = 'pd_op.gaussian'
|
|
self.set_parameter_op_name = 'builtin.set_parameter'
|
|
|
|
def get_operand_definition_op_attrs(self, cur_op, operand_name, attr_name):
|
|
input_names = cur_op.get_input_names()
|
|
self.assertIn(operand_name, input_names)
|
|
attr = (
|
|
cur_op.operand(input_names.index(operand_name))
|
|
.source()
|
|
.get_defining_op()
|
|
.attrs()[attr_name]
|
|
)
|
|
return attr
|
|
|
|
def get_init_ops_by_op_name(self, block, op_name):
|
|
checked_ops = []
|
|
for op in block.ops:
|
|
# get init op
|
|
if op_name == op.name():
|
|
checked_ops.append(op)
|
|
return checked_ops
|
|
|
|
def test_uniform_msra_initializer(self):
|
|
"""Test MSRA initializer with uniform distribution on
|
|
for matrix multiply.
|
|
"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.KaimingUniform(),
|
|
)
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_uniform_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
limit = np.sqrt(6.0 / param.shape[0])
|
|
min = self.get_operand_definition_op_attrs(
|
|
init_op, "min", "value"
|
|
)
|
|
max = self.get_operand_definition_op_attrs(
|
|
init_op, "max", "value"
|
|
)
|
|
self.assertAlmostEqual(min, -limit, delta=DELTA)
|
|
self.assertAlmostEqual(max, limit, delta=DELTA)
|
|
self.assertEqual(init_op.attrs()['seed'], 0)
|
|
|
|
def test_uniform_msra_initializer_conv(self):
|
|
"""Test MSRA initializer with uniform distribution on
|
|
for convolutions.
|
|
"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10, 15, 20],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.KaimingUniform(),
|
|
)
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_uniform_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
receptive_field_size = float(15 * 20)
|
|
limit = np.sqrt(6.0 / (param.shape[1] * receptive_field_size))
|
|
min = self.get_operand_definition_op_attrs(
|
|
init_op, "min", "value"
|
|
)
|
|
max = self.get_operand_definition_op_attrs(
|
|
init_op, "max", "value"
|
|
)
|
|
self.assertAlmostEqual(min, -limit, delta=DELTA)
|
|
self.assertAlmostEqual(max, limit, delta=DELTA)
|
|
self.assertEqual(init_op.attrs()['seed'], 0)
|
|
|
|
def test_normal_msra_initializer(self):
|
|
"""Test MSRA initializer with normal distribution on
|
|
for matrix multiply.
|
|
"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.KaimingNormal(),
|
|
)
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_normal_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
std = np.sqrt(2.0 / param.shape[0])
|
|
self.assertAlmostEqual(
|
|
init_op.attrs()['mean'], 0.0, delta=DELTA
|
|
)
|
|
self.assertAlmostEqual(init_op.attrs()['std'], std, delta=DELTA)
|
|
self.assertEqual(init_op.attrs()['seed'], 0)
|
|
|
|
def test_normal_msra_initializer_conv(self):
|
|
"""Test MSRA initializer with normal distribution on
|
|
for convolutions.
|
|
"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype="float32",
|
|
shape=[5, 10, 15, 20],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.KaimingNormal(),
|
|
)
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_normal_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
receptive_field_size = float(15 * 20)
|
|
std = np.sqrt(2.0 / (param.shape[1] * receptive_field_size))
|
|
self.assertAlmostEqual(
|
|
init_op.attrs()['mean'], 0.0, delta=DELTA
|
|
)
|
|
self.assertAlmostEqual(init_op.attrs()['std'], std, delta=DELTA)
|
|
self.assertEqual(init_op.attrs()['seed'], 0)
|
|
|
|
def test_msra_initializer_supplied_arguments(
|
|
self, dtype="float32", uniform=True
|
|
):
|
|
"""Test the MSRA initializer with supplied arguments"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype=dtype,
|
|
shape=[5, 10],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.MSRAInitializer(
|
|
fan_in=12, seed=134, uniform=uniform
|
|
),
|
|
)
|
|
block = startup.global_block()
|
|
init_op_name = (
|
|
self.init_uniform_op_name
|
|
if uniform
|
|
else self.init_normal_op_name
|
|
)
|
|
|
|
checked_ops = self.get_init_ops_by_op_name(block, init_op_name)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
init_op = checked_ops[0]
|
|
if uniform:
|
|
limit = np.sqrt(6.0 / 12)
|
|
min = self.get_operand_definition_op_attrs(
|
|
init_op, "min", "value"
|
|
)
|
|
max = self.get_operand_definition_op_attrs(
|
|
init_op, "max", "value"
|
|
)
|
|
self.assertAlmostEqual(min, -limit, delta=DELTA)
|
|
self.assertAlmostEqual(max, limit, delta=DELTA)
|
|
|
|
self.assertEqual(init_op.attrs()['seed'], 134)
|
|
|
|
return main, startup
|
|
|
|
@unittest.skipIf(
|
|
not (paddle.is_compiled_with_cuda() or is_custom_device()),
|
|
"core is not compiled with CUDA",
|
|
)
|
|
def test_msra_initializer_fp16(self):
|
|
"""Test the MSRA initializer with float16"""
|
|
main_1, startup_1 = self.test_msra_initializer_supplied_arguments(
|
|
"float16"
|
|
)
|
|
with paddle.pir_utils.IrGuard():
|
|
exe = paddle.static.Executor(get_device_place())
|
|
exe.run(startup_1)
|
|
exe.run(main_1)
|
|
|
|
main_2, startup_2 = self.test_msra_initializer_supplied_arguments(
|
|
"float16", uniform=False
|
|
)
|
|
with paddle.pir_utils.IrGuard():
|
|
exe = paddle.static.Executor(get_device_place())
|
|
exe.run(startup_2)
|
|
exe.run(main_2)
|
|
|
|
@unittest.skipIf(
|
|
not (paddle.base.core.is_compiled_with_cuda() or is_custom_device())
|
|
or not paddle.base.core.is_bfloat16_supported(get_device_place()),
|
|
"core is not compiled with CUDA and do not support bfloat16",
|
|
)
|
|
def test_msra_initializer_bf16(self):
|
|
"""Test the MSRA initializer with bfloat16"""
|
|
main_1, startup_1 = self.test_msra_initializer_supplied_arguments(
|
|
"uint16"
|
|
)
|
|
with paddle.pir_utils.IrGuard():
|
|
exe = paddle.static.Executor(get_device_place())
|
|
exe.run(startup_1)
|
|
exe.run(main_1)
|
|
|
|
main_2, startup_2 = self.test_msra_initializer_supplied_arguments(
|
|
"uint16", uniform=False
|
|
)
|
|
with paddle.pir_utils.IrGuard():
|
|
exe = paddle.static.Executor(get_device_place())
|
|
exe.run(startup_2)
|
|
exe.run(main_2)
|
|
|
|
|
|
class TestBilinearInitializer(unittest.TestCase):
|
|
def test_bilinear_initializer(self, dtype="float32"):
|
|
"""Test the bilinear initializer with supplied arguments"""
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype=dtype,
|
|
shape=[8, 1, 3, 3],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.Bilinear(),
|
|
)
|
|
num_ops = 2 if dtype in ["float16", "uint16", "float64"] else 1
|
|
self.assertEqual(len(block.ops), num_ops)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'assign_value')
|
|
return block
|
|
|
|
def test_bilinear_initializer_fp64(self):
|
|
self.test_bilinear_initializer(dtype='float64')
|
|
|
|
def test_bilinear_initializer_fp16(self):
|
|
"""Test the bilinear initializer with supplied arguments"""
|
|
block = self.test_bilinear_initializer("float16")
|
|
self.assertTrue(check_cast_op(block.ops[1]))
|
|
|
|
def test_bilinear_initializer_bf16(self):
|
|
"""Test the bilinear initializer with supplied arguments"""
|
|
block = self.test_bilinear_initializer("uint16")
|
|
self.assertTrue(check_cast_op(block.ops[1]))
|
|
|
|
def test_type_error(self):
|
|
self.assertRaises(TypeError, self.test_bilinear_initializer, 'int32')
|
|
|
|
|
|
class TestBilinearInitializerPir(unittest.TestCase):
|
|
def setUp(self):
|
|
self.set_parameter_op_name = 'builtin.set_parameter'
|
|
self.init_op_name = "pd_op.assign_value"
|
|
self.cast_op_name = "pd_op.cast"
|
|
|
|
def get_operand_definition_op_attrs(self, cur_op, operand_name, attr_name):
|
|
input_names = cur_op.get_input_names()
|
|
self.assertIn(operand_name, input_names)
|
|
attr = (
|
|
cur_op.operand(input_names.index(operand_name))
|
|
.source()
|
|
.get_defining_op()
|
|
.attrs()[attr_name]
|
|
)
|
|
return attr
|
|
|
|
def get_init_ops_by_op_name(self, block, op_name):
|
|
checked_ops = []
|
|
for op in block.ops:
|
|
# get init op
|
|
if op_name == op.name():
|
|
checked_ops.append(op)
|
|
return checked_ops
|
|
|
|
def test_bilinear_initializer(self, dtype="float32"):
|
|
"""Test the bilinear initializer with supplied arguments"""
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype=dtype,
|
|
shape=[8, 1, 3, 3],
|
|
name="param",
|
|
initializer=paddle.nn.initializer.Bilinear(),
|
|
)
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
checked_cast_ops = self.get_init_ops_by_op_name(
|
|
block, self.cast_op_name
|
|
)
|
|
num_cast_op = (
|
|
1 if dtype in ["float16", "uint16", "float64"] else 0
|
|
)
|
|
self.assertEqual(len(checked_cast_ops), num_cast_op)
|
|
|
|
return startup
|
|
|
|
def test_bilinear_initializer_fp64(self):
|
|
self.test_bilinear_initializer(dtype='float64')
|
|
|
|
def test_bilinear_initializer_fp16(self):
|
|
"""Test the bilinear initializer with supplied arguments"""
|
|
startup = self.test_bilinear_initializer("float16")
|
|
cast_ops = self.get_init_ops_by_op_name(
|
|
startup.global_block(), self.cast_op_name
|
|
)
|
|
self.assertGreater(len(cast_ops), 0)
|
|
cast_op = cast_ops[0]
|
|
self.assertTrue(check_cast_op_pir(cast_op))
|
|
|
|
def test_bilinear_initializer_bf16(self):
|
|
"""Test the bilinear initializer with supplied arguments"""
|
|
startup = self.test_bilinear_initializer("uint16")
|
|
cast_ops = self.get_init_ops_by_op_name(
|
|
startup.global_block(), self.cast_op_name
|
|
)
|
|
self.assertGreater(len(cast_ops), 0)
|
|
cast_op = cast_ops[0]
|
|
self.assertTrue(check_cast_op_pir(cast_op))
|
|
|
|
def test_type_error(self):
|
|
self.assertRaises(TypeError, self.test_bilinear_initializer, 'int32')
|
|
|
|
|
|
class TestBilinearInitializerDygraphAPI(unittest.TestCase):
|
|
def func_test_case(self):
|
|
factor = 2
|
|
C = 2
|
|
B = 8
|
|
H = W = 32
|
|
w_attr = paddle.ParamAttr(
|
|
learning_rate=0.0,
|
|
regularizer=L2Decay(0.0),
|
|
initializer=paddle.nn.initializer.Bilinear(),
|
|
)
|
|
data = paddle.rand([B, 3, H, W], dtype='float32')
|
|
conv_up = paddle.nn.Conv2DTranspose(
|
|
3,
|
|
out_channels=C,
|
|
kernel_size=2 * factor - factor % 2,
|
|
padding=int(math.ceil((factor - 1) / 2.0)),
|
|
stride=factor,
|
|
weight_attr=w_attr,
|
|
bias_attr=False,
|
|
)
|
|
x = conv_up(data)
|
|
return x
|
|
|
|
def func_test_case_fp16(self):
|
|
paddle.set_default_dtype("float16")
|
|
paddle.seed(1234)
|
|
w_attr = paddle.ParamAttr(
|
|
learning_rate=0.0,
|
|
regularizer=L2Decay(0.0),
|
|
initializer=paddle.nn.initializer.Bilinear(),
|
|
)
|
|
conv2d = paddle.nn.Conv2D(1, 2, 3, weight_attr=w_attr)
|
|
paddle.set_default_dtype("float32")
|
|
return conv2d.weight
|
|
|
|
def test_bilinear_initializer(self):
|
|
paddle.disable_static()
|
|
eager_x = self.func_test_case()
|
|
legacy_x = self.func_test_case()
|
|
self.assertEqual(eager_x.numpy().all(), legacy_x.numpy().all())
|
|
paddle.enable_static()
|
|
|
|
def test_bilinear_initializer_fp16(self):
|
|
paddle.disable_static()
|
|
eager_x = self.func_test_case_fp16()
|
|
legacy_x = self.func_test_case_fp16()
|
|
self.assertEqual(eager_x.numpy().all(), legacy_x.numpy().all())
|
|
paddle.enable_static()
|
|
|
|
|
|
class TestNumpyArrayInitializer(unittest.TestCase):
|
|
def test_numpy_array_initializer(self, dtype="float32"):
|
|
"""Test the numpy array initializer with supplied arguments"""
|
|
import numpy
|
|
|
|
with paddle.pir_utils.OldIrGuard():
|
|
program = framework.Program()
|
|
block = program.global_block()
|
|
np_array = numpy.random.random(10000).astype(dtype)
|
|
for _ in range(2):
|
|
block.create_parameter(
|
|
dtype=np_array.dtype,
|
|
shape=np_array.shape,
|
|
name="param",
|
|
initializer=paddle.nn.initializer.Assign(np_array),
|
|
)
|
|
num_ops = 2 if dtype in ["float16", "uint16"] else 1
|
|
self.assertEqual(len(block.ops), num_ops)
|
|
init_op = block.ops[0]
|
|
self.assertEqual(init_op.type, 'assign_value')
|
|
values = framework.extract_plain_list(init_op.attr('values'))
|
|
assert values == np_array.ravel().tolist()
|
|
return block
|
|
|
|
def test_numpy_array_initializer_fp16(self):
|
|
"""Test the numpy array initializer with float16"""
|
|
block = self.test_numpy_array_initializer("float16")
|
|
self.assertTrue(block.ops[1])
|
|
|
|
def test_numpy_array_initializer_bf16(self):
|
|
"""Test the numpy array initializer with bfloat16"""
|
|
block = self.test_numpy_array_initializer("uint16")
|
|
self.assertTrue(block.ops[1])
|
|
|
|
|
|
class TestNumpyArrayInitializerPir(unittest.TestCase):
|
|
def setUp(self):
|
|
self.set_parameter_op_name = 'builtin.set_parameter'
|
|
self.init_op_name = "pd_op.assign_value"
|
|
self.cast_op_name = "pd_op.cast"
|
|
|
|
def get_operand_definition_op_attrs(self, cur_op, operand_name, attr_name):
|
|
input_names = cur_op.get_input_names()
|
|
self.assertIn(operand_name, input_names)
|
|
attr = (
|
|
cur_op.operand(input_names.index(operand_name))
|
|
.source()
|
|
.get_defining_op()
|
|
.attrs()[attr_name]
|
|
)
|
|
return attr
|
|
|
|
def get_init_ops_by_op_name(self, block, op_name):
|
|
checked_ops = []
|
|
for op in block.ops:
|
|
# get init op
|
|
if op_name == op.name():
|
|
checked_ops.append(op)
|
|
return checked_ops
|
|
|
|
def test_numpy_array_initializer(self, dtype="float32"):
|
|
"""Test the numpy array initializer with supplied arguments"""
|
|
np_array = np.random.random(10000).astype(dtype)
|
|
with paddle.pir_utils.IrGuard():
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype=np_array.dtype,
|
|
shape=np_array.shape,
|
|
name="param",
|
|
initializer=paddle.nn.initializer.Assign(np_array),
|
|
)
|
|
block = startup.global_block()
|
|
checked_ops = self.get_init_ops_by_op_name(
|
|
block, self.init_op_name
|
|
)
|
|
self.assertEqual(len(checked_ops), 1)
|
|
checked_cast_ops = self.get_init_ops_by_op_name(
|
|
block, self.cast_op_name
|
|
)
|
|
num_cast_op = 1 if dtype in ["float16", "uint16"] else 0
|
|
self.assertEqual(len(checked_cast_ops), num_cast_op)
|
|
|
|
init_op = checked_ops[0]
|
|
assert (init_op.attrs()['values'] == np_array).all()
|
|
|
|
return startup
|
|
|
|
def test_numpy_array_initializer_fp16(self):
|
|
"""Test the numpy array initializer with float16"""
|
|
startup = self.test_numpy_array_initializer("float16")
|
|
cast_ops = self.get_init_ops_by_op_name(
|
|
startup.global_block(), self.cast_op_name
|
|
)
|
|
self.assertGreater(len(cast_ops), 0)
|
|
cast_op = cast_ops[0]
|
|
self.assertTrue(check_cast_op_pir(cast_op))
|
|
|
|
def test_numpy_array_initializer_bf16(self):
|
|
"""Test the numpy array initializer with bfloat16"""
|
|
startup = self.test_numpy_array_initializer("uint16")
|
|
cast_ops = self.get_init_ops_by_op_name(
|
|
startup.global_block(), self.cast_op_name
|
|
)
|
|
self.assertGreater(len(cast_ops), 0)
|
|
cast_op = cast_ops[0]
|
|
self.assertTrue(check_cast_op_pir(cast_op))
|
|
|
|
|
|
class TestUniformInitializerDygraph(unittest.TestCase):
|
|
def test_uniform_initializer(self, dtype="float32"):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
paddle.disable_static()
|
|
|
|
tensor = paddle.zeros([1024, 1024, 16])
|
|
tensor.stop_gradient = False
|
|
np.testing.assert_allclose(
|
|
np.zeros((1024, 1024, 16)), tensor.numpy(), rtol=1e-05
|
|
)
|
|
|
|
uniform_ = paddle.nn.initializer.Uniform()
|
|
uniform_(tensor)
|
|
|
|
self.assertEqual(
|
|
tensor.stop_gradient, False
|
|
) # stop_gradient is not changed
|
|
|
|
hist, prob = output_hist(tensor.numpy())
|
|
|
|
np.testing.assert_allclose(hist, prob, rtol=0, atol=0.001)
|
|
|
|
paddle.enable_static()
|
|
|
|
|
|
class TestXavierInitializerDygraph(unittest.TestCase):
|
|
def test_xavier_initializer(self, dtype="float32"):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
paddle.disable_static()
|
|
|
|
tensor = paddle.zeros([1024, 1024, 16])
|
|
tensor.stop_gradient = False
|
|
|
|
xavier_ = paddle.nn.initializer.XavierNormal(fan_in=3, fan_out=5)
|
|
xavier_(tensor)
|
|
|
|
hist, _ = output_hist(tensor.numpy())
|
|
|
|
hist2, _ = output_hist(
|
|
np.random.normal(0, np.sqrt(2.0 / (3 + 5)), [1024, 1024, 16])
|
|
)
|
|
|
|
np.testing.assert_allclose(hist, hist2, rtol=0, atol=0.01)
|
|
paddle.enable_static()
|
|
|
|
def test_xavier_normal_initializer_zero_size(self, dtype="float32"):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
paddle.disable_static()
|
|
|
|
tensor = paddle.zeros([0, 0, 0])
|
|
tensor.stop_gradient = False
|
|
|
|
xavier_ = paddle.nn.initializer.XavierNormal(fan_in=0, fan_out=0)
|
|
xavier_(tensor)
|
|
self.assertEqual(tensor.stop_gradient, False)
|
|
self.assertEqual(tensor.shape, [0, 0, 0])
|
|
|
|
paddle.enable_static()
|
|
|
|
def test_xavier_uniform_initializer_zero_size(self, dtype="float32"):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
paddle.disable_static()
|
|
|
|
tensor = paddle.zeros([0, 0, 0])
|
|
tensor.stop_gradient = False
|
|
|
|
xavier_ = paddle.nn.initializer.XavierUniform(fan_in=0, fan_out=0)
|
|
xavier_(tensor)
|
|
self.assertEqual(tensor.stop_gradient, False)
|
|
self.assertEqual(tensor.shape, [0, 0, 0])
|
|
|
|
paddle.enable_static()
|
|
|
|
|
|
class TestXavierInitializerDygraph2(unittest.TestCase):
|
|
def test_xavier_initializer_with_gain(self, dtype="float32"):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
paddle.disable_static()
|
|
|
|
tensor = paddle.zeros([1024, 1024, 16])
|
|
tensor.stop_gradient = False
|
|
|
|
xavier_ = paddle.nn.initializer.XavierNormal(
|
|
fan_in=3, fan_out=5, gain=2.5
|
|
)
|
|
xavier_(tensor)
|
|
|
|
hist, _ = output_hist(tensor.numpy())
|
|
|
|
hist2, _ = output_hist(
|
|
np.random.normal(0, 2.5 * np.sqrt(2.0 / (3 + 5)), [1024, 1024, 16])
|
|
)
|
|
|
|
np.testing.assert_allclose(hist, hist2, rtol=0, atol=0.01)
|
|
paddle.enable_static()
|
|
|
|
|
|
class TestMSRAInitializerDygraph(unittest.TestCase):
|
|
def test_msra_initializer(self, dtype="float32"):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
paddle.disable_static()
|
|
|
|
tensor = paddle.zeros([1024, 1024, 16])
|
|
tensor.stop_gradient = False
|
|
|
|
msra_ = paddle.nn.initializer.KaimingNormal(fan_in=4)
|
|
msra_(tensor)
|
|
|
|
hist, _ = output_hist(tensor.numpy())
|
|
|
|
hist2, _ = output_hist(
|
|
np.random.normal(0, np.sqrt(2.0 / (4)), [1024, 1024, 16])
|
|
)
|
|
|
|
np.testing.assert_allclose(hist, hist2, rtol=0, atol=0.01)
|
|
paddle.enable_static()
|
|
|
|
|
|
class TestMSRAInitializerFanoutDygraph(unittest.TestCase):
|
|
def test_msra_fanout_initializer(self, dtype="float32"):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
paddle.disable_static()
|
|
|
|
tensor = paddle.zeros([16, 1024])
|
|
tensor.stop_gradient = False
|
|
|
|
msra_ = paddle.nn.initializer.KaimingNormal(mode='fan_out')
|
|
msra_(tensor)
|
|
|
|
hist, _ = output_hist(tensor.numpy())
|
|
|
|
hist2, _ = output_hist(
|
|
np.random.normal(0, np.sqrt(2.0 / (1024)), [16, 1024])
|
|
)
|
|
|
|
np.testing.assert_allclose(hist, hist2, rtol=0, atol=0.01)
|
|
paddle.enable_static()
|
|
|
|
def test_msra_invalid_fanout_initializer(self, dtype="float32"):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
paddle.disable_static()
|
|
|
|
tensor = paddle.zeros([16, 1024])
|
|
tensor.stop_gradient = False
|
|
|
|
with self.assertRaises(ValueError):
|
|
msra_ = paddle.nn.initializer.KaimingNormal(mode='fan')
|
|
msra_(tensor)
|
|
|
|
with self.assertRaises(ValueError):
|
|
msra_ = paddle.nn.initializer.KaimingNormal(
|
|
fan_in=1, mode='fan_out'
|
|
)
|
|
msra_(tensor)
|
|
|
|
def test_msra_uniform_fanout_initializer(self, dtype="float32"):
|
|
paddle.disable_static()
|
|
|
|
tensor = paddle.zeros([16, 1024])
|
|
tensor.stop_gradient = False
|
|
|
|
msra_ = paddle.nn.initializer.KaimingUniform(mode='fan_out')
|
|
msra_(tensor)
|
|
|
|
hist, _ = output_hist(tensor.numpy())
|
|
|
|
fan_out = tensor.shape[1]
|
|
limit = np.sqrt(6.0 / fan_out)
|
|
theory_data = np.random.uniform(-limit, limit, [16, 1024])
|
|
|
|
hist2, _ = output_hist(theory_data)
|
|
|
|
np.testing.assert_allclose(hist, hist2, rtol=0, atol=0.01)
|
|
paddle.enable_static()
|
|
|
|
|
|
class TestConsistencyOfDynamicAndStaticGraph(unittest.TestCase):
|
|
def test_order(self):
|
|
paddle.set_device('cpu')
|
|
SEED = 123
|
|
weight_attr = paddle.framework.ParamAttr(
|
|
name="linear_weight2",
|
|
learning_rate=1.0,
|
|
trainable=False,
|
|
regularizer=None,
|
|
initializer=paddle.nn.initializer.TruncatedNormal(
|
|
mean=0.0, std=2.0
|
|
),
|
|
)
|
|
bias_attr = paddle.framework.ParamAttr(
|
|
name="linear_bias2",
|
|
learning_rate=1.0,
|
|
trainable=False,
|
|
regularizer=None,
|
|
initializer=paddle.nn.initializer.TruncatedNormal(
|
|
mean=0.0, std=2.0
|
|
),
|
|
)
|
|
|
|
def run_dynamic_graph():
|
|
paddle.seed(SEED)
|
|
linear = paddle.nn.Linear(
|
|
1,
|
|
1,
|
|
weight_attr=paddle.framework.ParamAttr(
|
|
name="linear_weight1",
|
|
learning_rate=1.0,
|
|
trainable=False,
|
|
regularizer=None,
|
|
initializer=paddle.nn.initializer.TruncatedNormal(
|
|
mean=0.0, std=2.0
|
|
),
|
|
),
|
|
bias_attr=paddle.framework.ParamAttr(
|
|
name="linear_bias1",
|
|
learning_rate=1.0,
|
|
trainable=False,
|
|
regularizer=None,
|
|
initializer=paddle.nn.initializer.TruncatedNormal(
|
|
mean=0.0, std=2.0
|
|
),
|
|
),
|
|
)
|
|
return linear.weight.numpy(), linear.bias.numpy()
|
|
|
|
def run_static_graph():
|
|
exe = paddle.static.Executor(paddle.CPUPlace())
|
|
paddle.seed(SEED)
|
|
linear = paddle.nn.Linear(
|
|
1, 1, weight_attr=weight_attr, bias_attr=bias_attr
|
|
)
|
|
res = exe.run(
|
|
paddle.static.default_startup_program(),
|
|
fetch_list=[linear.weight, linear.bias],
|
|
)
|
|
return res[0], res[1]
|
|
|
|
with dygraph_guard():
|
|
dynamic_res = run_dynamic_graph()
|
|
with static_guard():
|
|
static_res = run_static_graph()
|
|
|
|
np.testing.assert_array_equal(dynamic_res[0], static_res[0])
|
|
np.testing.assert_array_equal(dynamic_res[1], static_res[1])
|
|
|
|
def test_assign_static_fp32(self):
|
|
random_value = np.random.randn(128, 128).astype("float32")
|
|
|
|
def run_dynamic_graph(dtype):
|
|
with dygraph_guard():
|
|
w = paddle.create_parameter(
|
|
random_value.shape,
|
|
dtype,
|
|
default_initializer=paddle.nn.initializer.Assign(
|
|
random_value
|
|
),
|
|
)
|
|
return w
|
|
|
|
def run_static_graph(dtype):
|
|
with static_guard():
|
|
exe = paddle.static.Executor(paddle.CPUPlace())
|
|
w = paddle.create_parameter(
|
|
random_value.shape,
|
|
dtype,
|
|
"w",
|
|
default_initializer=paddle.nn.initializer.Assign(
|
|
random_value
|
|
),
|
|
)
|
|
res = exe.run(
|
|
paddle.static.default_startup_program(),
|
|
fetch_list=w,
|
|
)
|
|
return res[0]
|
|
|
|
def run_pir_graph(dtype):
|
|
with paddle.pir_utils.IrGuard():
|
|
exe = paddle.static.Executor(paddle.CPUPlace())
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype=dtype,
|
|
shape=random_value.shape,
|
|
name="w",
|
|
initializer=paddle.nn.initializer.Assign(random_value),
|
|
)
|
|
exe.run(startup)
|
|
res = exe.run(main, fetch_list=[param])
|
|
return res[0]
|
|
|
|
dynamic_res = run_dynamic_graph("float32")
|
|
static_res = run_static_graph("float32")
|
|
pir_res = run_pir_graph("float32")
|
|
|
|
np.testing.assert_array_equal(dynamic_res.numpy(), static_res)
|
|
np.testing.assert_array_equal(dynamic_res.numpy(), pir_res)
|
|
|
|
def test_assign_static_fp64(self):
|
|
random_value = np.random.randn(128, 128).astype("float64")
|
|
|
|
def run_dynamic_graph(dtype):
|
|
with dygraph_guard():
|
|
w = paddle.create_parameter(
|
|
random_value.shape,
|
|
dtype,
|
|
"www",
|
|
default_initializer=paddle.nn.initializer.Assign(
|
|
random_value
|
|
),
|
|
)
|
|
return w
|
|
|
|
def run_static_graph(dtype):
|
|
with static_guard():
|
|
exe = paddle.static.Executor(paddle.CPUPlace())
|
|
w = paddle.create_parameter(
|
|
random_value.shape,
|
|
dtype,
|
|
"ww",
|
|
default_initializer=paddle.nn.initializer.Assign(
|
|
random_value
|
|
),
|
|
)
|
|
res = exe.run(
|
|
paddle.static.default_startup_program(),
|
|
fetch_list=w,
|
|
)
|
|
return res[0]
|
|
|
|
def run_pir_graph(dtype):
|
|
with paddle.pir_utils.IrGuard():
|
|
exe = paddle.static.Executor(paddle.CPUPlace())
|
|
main = paddle.static.Program()
|
|
startup = paddle.static.Program()
|
|
with paddle.static.program_guard(main, startup):
|
|
param = paddle.pir.core.create_parameter(
|
|
dtype=dtype,
|
|
shape=random_value.shape,
|
|
name="ww",
|
|
initializer=paddle.nn.initializer.Assign(random_value),
|
|
)
|
|
exe.run(startup)
|
|
res = exe.run(main, fetch_list=[param])
|
|
return res[0]
|
|
|
|
dynamic_res = run_dynamic_graph("float64")
|
|
static_res = run_static_graph("float64")
|
|
pir_res = run_pir_graph("float64")
|
|
|
|
np.testing.assert_array_equal(dynamic_res.numpy(), static_res)
|
|
np.testing.assert_array_equal(dynamic_res.numpy(), pir_res)
|
|
|
|
|
|
# 2-D Parameter with shape: [10, 15]
|
|
class TestOrthogonalInitializer1(unittest.TestCase):
|
|
"""
|
|
case 1
|
|
"""
|
|
|
|
def config(self):
|
|
self.weight_attr = paddle.ParamAttr(
|
|
initializer=paddle.nn.initializer.Orthogonal(gain=3.0)
|
|
)
|
|
self.dtype = "float64"
|
|
self.in_features = 10
|
|
self.out_features = 15
|
|
self.num_ops = 9
|
|
|
|
def check_result(self, a, b):
|
|
np.testing.assert_array_equal(a, b)
|
|
np.testing.assert_allclose(
|
|
np.matmul(a, a.T), 9 * np.eye(10), rtol=1e-5, atol=1e-8
|
|
)
|
|
|
|
def test_orthogonal(self):
|
|
self.config()
|
|
paddle.set_default_dtype(self.dtype)
|
|
|
|
paddle.disable_static()
|
|
paddle.seed(2021)
|
|
linear = paddle.nn.Linear(
|
|
self.in_features, self.out_features, weight_attr=self.weight_attr
|
|
)
|
|
res_dygraph = linear.weight.numpy()
|
|
|
|
paddle.enable_static()
|
|
paddle.seed(2021)
|
|
start_prog = paddle.static.Program()
|
|
main_prog = paddle.static.Program()
|
|
with paddle.static.program_guard(main_prog, start_prog):
|
|
linear = paddle.nn.Linear(
|
|
self.in_features,
|
|
self.out_features,
|
|
weight_attr=self.weight_attr,
|
|
)
|
|
|
|
block = start_prog.global_block()
|
|
if not paddle.framework.use_pir_api():
|
|
self.assertEqual(len(block.ops), self.num_ops)
|
|
self.assertEqual(block.ops[0].type, 'gaussian_random')
|
|
self.assertEqual(block.ops[1].type, 'qr')
|
|
self.assertEqual(block.ops[2].type, 'diag_v2')
|
|
self.assertEqual(block.ops[3].type, 'sign')
|
|
self.assertEqual(block.ops[4].type, 'elementwise_mul')
|
|
self.assertEqual(block.ops[-3].type, 'reshape2')
|
|
self.assertEqual(block.ops[-2].type, 'scale')
|
|
|
|
exe = paddle.static.Executor()
|
|
res_static = exe.run(start_prog, fetch_list=[linear.weight])[0]
|
|
|
|
self.check_result(res_dygraph, res_static)
|
|
|
|
def test_orthogonal_pir(self):
|
|
self.config()
|
|
paddle.set_default_dtype(self.dtype)
|
|
|
|
paddle.disable_static()
|
|
paddle.seed(2021)
|
|
linear = paddle.nn.Linear(
|
|
self.in_features, self.out_features, weight_attr=self.weight_attr
|
|
)
|
|
res_dygraph = linear.weight.numpy()
|
|
|
|
paddle.enable_static()
|
|
paddle.seed(2021)
|
|
start_prog = paddle.static.Program()
|
|
main_prog = paddle.static.Program()
|
|
with paddle.static.program_guard(main_prog, start_prog):
|
|
linear = paddle.nn.Linear(
|
|
self.in_features,
|
|
self.out_features,
|
|
weight_attr=self.weight_attr,
|
|
)
|
|
|
|
exe = paddle.static.Executor()
|
|
res_static = exe.run(start_prog, fetch_list=[linear.weight])[0]
|
|
|
|
self.check_result(res_dygraph, res_static)
|
|
|
|
|
|
# 2-D Parameter with shape: [15, 10]
|
|
class TestOrthogonalInitializer2(TestOrthogonalInitializer1):
|
|
"""
|
|
case 2
|
|
"""
|
|
|
|
def config(self):
|
|
self.weight_attr = paddle.ParamAttr(
|
|
initializer=paddle.nn.initializer.Orthogonal(gain=2.0)
|
|
)
|
|
self.dtype = "float64"
|
|
self.in_features = 15
|
|
self.out_features = 10
|
|
self.num_ops = 8
|
|
|
|
def check_result(self, a, b):
|
|
np.testing.assert_array_equal(a, b)
|
|
np.testing.assert_allclose(
|
|
np.matmul(a.T, a), 4 * np.eye(10), rtol=1e-5, atol=1e-8
|
|
)
|
|
|
|
|
|
# 2-D Parameter with shape: [10, 10]
|
|
class TestOrthogonalInitializer3(TestOrthogonalInitializer1):
|
|
"""
|
|
case 3
|
|
"""
|
|
|
|
def config(self):
|
|
self.weight_attr = paddle.ParamAttr(
|
|
initializer=paddle.nn.initializer.Orthogonal()
|
|
)
|
|
self.dtype = "float32"
|
|
self.in_features = 10
|
|
self.out_features = 10
|
|
self.num_ops = 8
|
|
|
|
def check_result(self, a, b):
|
|
np.testing.assert_array_equal(a, b)
|
|
np.testing.assert_allclose(
|
|
np.matmul(a.T, a), np.eye(10), rtol=1e-05, atol=1e-06
|
|
)
|
|
np.testing.assert_allclose(
|
|
np.matmul(a, a.T), np.eye(10), rtol=1e-05, atol=1e-06
|
|
)
|
|
|
|
def test_error(self):
|
|
self.config()
|
|
with self.assertRaises(AssertionError):
|
|
paddle.nn.Linear(10, 10, bias_attr=self.weight_attr)
|
|
|
|
|
|
# 4-D Parameter with shape: [6, 4, 3, 3]
|
|
class TestOrthogonalInitializer4(unittest.TestCase):
|
|
"""
|
|
case 4
|
|
"""
|
|
|
|
def config(self):
|
|
self.weight_attr = paddle.ParamAttr(
|
|
initializer=paddle.nn.initializer.Orthogonal(gain=3.0)
|
|
)
|
|
self.dtype = "float64"
|
|
self.in_features = 4
|
|
self.out_features = 6
|
|
self.kernel_size = (3, 3)
|
|
|
|
def check_result(self, a, b):
|
|
np.testing.assert_array_equal(a, b)
|
|
a = a.reshape(6, -1)
|
|
np.testing.assert_allclose(
|
|
np.matmul(a, a.T), 9 * np.eye(6), rtol=1e-5, atol=1e-8
|
|
)
|
|
|
|
def test_orthogonal(self):
|
|
self.config()
|
|
paddle.set_default_dtype(self.dtype)
|
|
|
|
paddle.disable_static()
|
|
paddle.seed(2021)
|
|
conv2d = paddle.nn.Conv2D(
|
|
self.in_features,
|
|
self.out_features,
|
|
self.kernel_size,
|
|
weight_attr=self.weight_attr,
|
|
)
|
|
res_dygraph = conv2d.weight.numpy()
|
|
|
|
paddle.enable_static()
|
|
paddle.seed(2021)
|
|
start_prog = paddle.static.Program()
|
|
main_prog = paddle.static.Program()
|
|
with paddle.static.program_guard(main_prog, start_prog):
|
|
inp = paddle.rand(shape=[8, self.in_features, 10, 10])
|
|
conv2d = paddle.nn.Conv2D(
|
|
self.in_features,
|
|
self.out_features,
|
|
self.kernel_size,
|
|
weight_attr=self.weight_attr,
|
|
)
|
|
output = conv2d(inp)
|
|
exe = paddle.static.Executor()
|
|
|
|
exe.run(start_prog)
|
|
res_static = exe.run(main_prog, fetch_list=[conv2d.weight])[0]
|
|
self.check_result(res_dygraph, res_static)
|
|
|
|
|
|
# 4-D Parameter with shape: [50, 4, 3, 3]
|
|
class TestOrthogonalInitializer5(TestOrthogonalInitializer4):
|
|
"""
|
|
case 5
|
|
"""
|
|
|
|
def config(self):
|
|
self.weight_attr = paddle.ParamAttr(
|
|
initializer=paddle.nn.initializer.Orthogonal(gain=2.0)
|
|
)
|
|
self.dtype = "float64"
|
|
self.in_features = 4
|
|
self.out_features = 50
|
|
self.kernel_size = (3, 3)
|
|
|
|
def check_result(self, a, b):
|
|
np.testing.assert_array_equal(a, b)
|
|
a = a.reshape(50, -1)
|
|
np.testing.assert_allclose(
|
|
np.matmul(a.T, a), 4 * np.eye(36), rtol=1e-5, atol=1e-8
|
|
)
|
|
|
|
|
|
# 4-D Parameter with shape: [36, 4, 3, 3]
|
|
class TestOrthogonalInitializer6(TestOrthogonalInitializer4):
|
|
"""
|
|
case 6
|
|
"""
|
|
|
|
def config(self):
|
|
self.weight_attr = paddle.ParamAttr(
|
|
initializer=paddle.nn.initializer.Orthogonal()
|
|
)
|
|
self.dtype = "float32"
|
|
self.in_features = 4
|
|
self.out_features = 36
|
|
self.kernel_size = (3, 3)
|
|
|
|
def check_result(self, a, b):
|
|
np.testing.assert_array_equal(a, b)
|
|
a = a.reshape(36, -1)
|
|
np.testing.assert_allclose(
|
|
np.matmul(a.T, a), np.eye(36), rtol=1e-05, atol=1e-06
|
|
)
|
|
np.testing.assert_allclose(
|
|
np.matmul(a, a.T), np.eye(36), rtol=1e-05, atol=1e-06
|
|
)
|
|
|
|
|
|
# initialize Conv1D weight
|
|
class TestDiracInitializer1(unittest.TestCase):
|
|
def config(self):
|
|
self.weight_attr = paddle.ParamAttr(
|
|
initializer=paddle.nn.initializer.Dirac()
|
|
)
|
|
self.dtype = "float64"
|
|
self.in_channels = 3
|
|
self.out_channels = 2
|
|
self.kernel_size = 3
|
|
self.input_shape = [8, self.in_channels, 10]
|
|
self.conv_layer = paddle.nn.Conv1D
|
|
self.num_ops = (
|
|
8 # fill_constant*2, reshape*2, assign_value*2, scatter, cast
|
|
)
|
|
|
|
def check_result(self, w_dygraph, w_static, conv_in, conv_out):
|
|
np.testing.assert_array_equal(w_dygraph, w_static)
|
|
np.testing.assert_array_equal(conv_out, conv_in[:, 0:2, 1:9])
|
|
|
|
def test_dirac(self):
|
|
self.config()
|
|
paddle.set_default_dtype(self.dtype)
|
|
|
|
paddle.disable_static()
|
|
conv = self.conv_layer(
|
|
self.in_channels,
|
|
self.out_channels,
|
|
self.kernel_size,
|
|
weight_attr=self.weight_attr,
|
|
)
|
|
weight_dygraph = conv.weight.numpy()
|
|
|
|
paddle.enable_static()
|
|
start_prog = paddle.static.Program()
|
|
main_prog = paddle.static.Program()
|
|
with paddle.static.program_guard(main_prog, start_prog):
|
|
inp = paddle.rand(self.input_shape)
|
|
conv = self.conv_layer(
|
|
self.in_channels,
|
|
self.out_channels,
|
|
self.kernel_size,
|
|
weight_attr=self.weight_attr,
|
|
)
|
|
|
|
output = conv(inp)
|
|
block = start_prog.global_block()
|
|
if not paddle.framework.use_pir_api():
|
|
self.assertEqual(len(block.ops), self.num_ops)
|
|
self.assertEqual(block.ops[0].type, 'fill_constant')
|
|
self.assertEqual(block.ops[1].type, 'reshape2')
|
|
self.assertEqual(block.ops[2].type, 'assign_value')
|
|
self.assertEqual(block.ops[3].type, 'assign_value')
|
|
self.assertEqual(block.ops[4].type, 'scatter')
|
|
self.assertEqual(block.ops[5].type, 'reshape2')
|
|
|
|
exe = paddle.static.Executor()
|
|
exe.run(start_prog)
|
|
fetch = exe.run(main_prog, fetch_list=[inp, output, conv.weight])
|
|
conv_input = fetch[0]
|
|
conv_output = fetch[1]
|
|
weight_static = fetch[2]
|
|
|
|
self.check_result(
|
|
weight_dygraph, weight_static, conv_input, conv_output
|
|
)
|
|
|
|
def test_dirac_pir(self):
|
|
self.config()
|
|
paddle.set_default_dtype(self.dtype)
|
|
|
|
paddle.disable_static()
|
|
conv = self.conv_layer(
|
|
self.in_channels,
|
|
self.out_channels,
|
|
self.kernel_size,
|
|
weight_attr=self.weight_attr,
|
|
)
|
|
weight_dygraph = conv.weight.numpy()
|
|
|
|
paddle.enable_static()
|
|
with paddle.pir_utils.IrGuard():
|
|
start_prog = paddle.static.Program()
|
|
main_prog = paddle.static.Program()
|
|
with paddle.static.program_guard(main_prog, start_prog):
|
|
inp = paddle.rand(self.input_shape)
|
|
conv = self.conv_layer(
|
|
self.in_channels,
|
|
self.out_channels,
|
|
self.kernel_size,
|
|
weight_attr=self.weight_attr,
|
|
)
|
|
|
|
output = conv(inp)
|
|
|
|
exe = paddle.static.Executor()
|
|
exe.run(start_prog)
|
|
fetch = exe.run(
|
|
main_prog, fetch_list=[inp, output, conv.weight]
|
|
)
|
|
conv_input = fetch[0]
|
|
conv_output = fetch[1]
|
|
weight_static = fetch[2]
|
|
|
|
self.check_result(
|
|
weight_dygraph, weight_static, conv_input, conv_output
|
|
)
|
|
|
|
|
|
# initialize Conv2D weight
|
|
class TestDiracInitializer2(TestDiracInitializer1):
|
|
def config(self):
|
|
self.weight_attr = paddle.ParamAttr(
|
|
initializer=paddle.nn.initializer.Dirac(groups=1)
|
|
)
|
|
self.dtype = "float64"
|
|
self.in_channels = 4
|
|
self.out_channels = 8
|
|
self.kernel_size = (3, 3)
|
|
self.input_shape = [8, self.in_channels, 10, 10]
|
|
self.conv_layer = paddle.nn.Conv2D
|
|
self.num_ops = 8
|
|
|
|
def check_result(self, w_dygraph, w_static, conv_in, conv_out):
|
|
np.testing.assert_array_equal(w_dygraph, w_static)
|
|
np.testing.assert_array_equal(
|
|
conv_out[:, 0:4, :, :], conv_in[:, :, 1:9, 1:9]
|
|
)
|
|
np.testing.assert_array_equal(
|
|
conv_out[:, 4:8, :, :], np.zeros([8, 4, 8, 8])
|
|
)
|
|
|
|
|
|
# initialize Conv3D weight
|
|
class TestDiracInitializer3(TestDiracInitializer1):
|
|
def config(self):
|
|
self.weight_attr = paddle.ParamAttr(
|
|
initializer=paddle.nn.initializer.Dirac(groups=2)
|
|
)
|
|
self.dtype = "float32"
|
|
self.in_channels = 5
|
|
self.out_channels = 10
|
|
self.kernel_size = (3, 3, 3)
|
|
self.input_shape = [8, self.in_channels, 10, 10, 10]
|
|
self.conv_layer = paddle.nn.Conv3D
|
|
self.num_ops = 7
|
|
|
|
def check_result(self, w_dygraph, w_static, conv_in, conv_out):
|
|
np.testing.assert_array_equal(w_dygraph, w_static)
|
|
np.testing.assert_array_equal(
|
|
conv_out[:, 0:5, :, :, :], conv_in[:, :, 1:9, 1:9, 1:9]
|
|
)
|
|
np.testing.assert_array_equal(
|
|
conv_out[:, 5:10, :, :, :], conv_in[:, :, 1:9, 1:9, 1:9]
|
|
)
|
|
|
|
def test_error(self):
|
|
self.config()
|
|
with self.assertRaises(AssertionError):
|
|
paddle.nn.Linear(10, 10, weight_attr=self.weight_attr)
|
|
|
|
with self.assertRaises(AssertionError):
|
|
paddle.nn.Conv2D(5, 9, (3, 3), weight_attr=self.weight_attr)
|
|
|
|
|
|
class TestTruncatedNormalInitializerDygraph(unittest.TestCase):
|
|
def _trunc_normal_numpy(self, tensor, mean=0.0, std=1.0, a=-2.0, b=2.0):
|
|
# Method based on https://people.sc.fsu.edu/~jburkardt/presentations/truncated_normal.pdf
|
|
def norm_cdf(x):
|
|
# Computes standard normal cumulative distribution function
|
|
return (1.0 + math.erf(x / math.sqrt(2.0))) / 2.0
|
|
|
|
# Values are generated by using a truncated uniform distribution and
|
|
# then using the inverse CDF for the normal distribution.
|
|
# Get upper and lower cdf values
|
|
l = norm_cdf((a - mean) / std)
|
|
u = norm_cdf((b - mean) / std)
|
|
|
|
# Uniformly fill tensor with values from [l, u], then translate to
|
|
# [2l-1, 2u-1].
|
|
_tensor = np.random.uniform(
|
|
low=2 * l - 1, high=2 * u - 1, size=tensor.shape
|
|
).astype(paddle.get_default_dtype())
|
|
|
|
# Use inverse cdf transform for normal distribution to get truncated
|
|
# standard normal
|
|
_tensor = special.erfinv(_tensor)
|
|
|
|
# Transform to proper mean, std
|
|
_tensor = np.multiply(_tensor, std * math.sqrt(2.0))
|
|
_tensor = np.add(_tensor, mean)
|
|
|
|
# Clamp to ensure it"s in the proper range
|
|
_tensor = np.clip(_tensor, a_min=a, a_max=b)
|
|
return _tensor
|
|
|
|
def test_truncated_normal_initializer_fp32(self):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
with dygraph_guard():
|
|
paddle.seed(42)
|
|
pre_dtype = paddle.get_default_dtype()
|
|
paddle.set_default_dtype("float32")
|
|
|
|
tensor = paddle.zeros([1024, 1024, 8])
|
|
tensor.stop_gradient = False
|
|
|
|
truncated_normal_ = paddle.nn.initializer.TruncatedNormal()
|
|
truncated_normal_(tensor)
|
|
|
|
array = self._trunc_normal_numpy(tensor)
|
|
np.testing.assert_allclose(
|
|
array.mean(), tensor.mean().item(), rtol=0.01, atol=0.01
|
|
)
|
|
np.testing.assert_allclose(
|
|
array.std(), tensor.std().item(), rtol=0.01, atol=0.01
|
|
)
|
|
paddle.set_default_dtype(pre_dtype)
|
|
|
|
def test_truncated_normal_initializer_fp64(self):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
with dygraph_guard():
|
|
paddle.seed(42)
|
|
pre_dtype = paddle.get_default_dtype()
|
|
paddle.set_default_dtype("float64")
|
|
|
|
tensor = paddle.zeros([1024, 1024, 8])
|
|
tensor.stop_gradient = False
|
|
|
|
truncated_normal_ = paddle.nn.initializer.TruncatedNormal()
|
|
truncated_normal_(tensor)
|
|
|
|
array = self._trunc_normal_numpy(tensor)
|
|
np.testing.assert_allclose(
|
|
array.mean(), tensor.mean().item(), rtol=0.01, atol=0.01
|
|
)
|
|
np.testing.assert_allclose(
|
|
array.std(), tensor.std().item(), rtol=0.01, atol=0.01
|
|
)
|
|
paddle.set_default_dtype(pre_dtype)
|
|
|
|
|
|
class TestAssignInitializerDygraph(unittest.TestCase):
|
|
def test_assign_initializer_fp32(self):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
with dygraph_guard():
|
|
pre_dtype = paddle.get_default_dtype()
|
|
paddle.set_default_dtype("float32")
|
|
|
|
tensor = paddle.zeros(
|
|
[1024, 1024, 8], dtype=paddle.get_default_dtype()
|
|
)
|
|
tensor.stop_gradient = False
|
|
array = np.random.randn(*tensor.shape).astype(
|
|
paddle.get_default_dtype()
|
|
)
|
|
|
|
assign_ = paddle.nn.initializer.Assign(array)
|
|
assign_(tensor)
|
|
|
|
np.testing.assert_allclose(array, tensor, rtol=1e-6, atol=1e-6)
|
|
paddle.set_default_dtype(pre_dtype)
|
|
|
|
def test_assign_initializer_fp64(self):
|
|
"""
|
|
In dygraph mode, we can use initializer directly to initialize a tensor.
|
|
"""
|
|
with dygraph_guard():
|
|
pre_dtype = paddle.get_default_dtype()
|
|
paddle.set_default_dtype("float64")
|
|
|
|
tensor = paddle.zeros(
|
|
[1024, 1024, 8], dtype=paddle.get_default_dtype()
|
|
)
|
|
tensor.stop_gradient = False
|
|
array = np.random.randn(*tensor.shape).astype(
|
|
paddle.get_default_dtype()
|
|
)
|
|
|
|
assign_ = paddle.nn.initializer.Assign(array)
|
|
assign_(tensor)
|
|
|
|
np.testing.assert_allclose(array, tensor, rtol=1e-6, atol=1e-6)
|
|
paddle.set_default_dtype(pre_dtype)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
paddle.enable_static()
|
|
unittest.main()
|