code: #   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.
  
  from __future__ import print_function
  import unittest
  import sys
  import numpy as np
  
  sys.path.append("..")
  from op_test import OpTest
  from op_test_xpu import XPUOpTest
  import paddle.fluid as fluid
  from paddle.fluid import Program, program_guard
  import paddle
  from xpu.get_test_cover_info import create_test_class, get_xpu_op_support_types, XPUOpTestWrapper
  
  paddle.enable_static()
  np.random.seed(10)
  
  
  # CANN Op Support X: float32, int32, int64
  # Situation 1: shape is a list(without tensor)
  class XPUTestExpandV2Op(XPUOpTestWrapper):
  
      def __init__(self):
          self.op_name = 'expand_v2'
          self.use_dynamic_create_class = False
  
      class TestExpandV2XPUOp(XPUOpTest):
  
          def setUp(self):
              self.init_dtype()
              self.set_xpu()
              self.op_type = "expand_v2"
              self.place = paddle.XPUPlace(0)
              self.init_data()
              self.inputs = {
                  'X': np.random.random(self.ori_shape).astype(self.dtype)
              }
              self.attrs = {'shape': self.shape}
              output = np.tile(self.inputs['X'], self.expand_times)
              self.outputs = {'Out': output}
  
          def init_dtype(self):
              self.dtype = self.in_type
  
          def set_xpu(self):
              self.__class__.use_xpu = True
              self.__class__.no_need_check_grad = True
  
          def init_data(self):
              self.ori_shape = [100]
              self.shape = [100]
              self.expand_times = [1]
  
          def test_check_output(self):
              self.check_output_with_place(self.place)
  
      class TestExpandV2OpRank2_DimExpanding(TestExpandV2XPUOp):
  
          def init_data(self):
              self.ori_shape = [120]
              self.shape = [2, 120]
              self.expand_times = [2, 1]
  
      class TestExpandV2OpRank2(TestExpandV2XPUOp):
  
          def init_data(self):
              self.ori_shape = [1, 140]
              self.shape = [12, 140]
              self.expand_times = [12, 1]
  
      class TestExpandV2OpRank3_Corner(TestExpandV2XPUOp):
  
          def init_data(self):
              self.ori_shape = (2, 10, 5)
              self.shape = (2, 10, 5)
              self.expand_times = (1, 1, 1)
  
      class TestExpandV2OpRank4(TestExpandV2XPUOp):
  
          def init_data(self):
              self.ori_shape = (2, 4, 5, 7)
              self.shape = (-1, -1, -1, -1)
              self.expand_times = (1, 1, 1, 1)
  
      class TestExpandV2OpRank5(TestExpandV2XPUOp):
  
          def init_data(self):
              self.ori_shape = (2, 4, 1, 15)
              self.shape = (2, -1, 4, -1)
              self.expand_times = (1, 1, 4, 1)
  
      class TestExpandV2OpRank6(TestExpandV2XPUOp):
  
          def init_data(self):
              self.ori_shape = (4, 1, 30)
              self.shape = (2, -1, 4, 30)
              self.expand_times = (2, 1, 4, 1)
  
      # Situation 2: shape is a list(with tensor)
      class TestExpandV2OpXPURank1_tensor_attr(TestExpandV2XPUOp):
  
          def setUp(self):
              self.set_xpu()
              self.place = paddle.XPUPlace(0)
              self.op_type = "expand_v2"
              self.init_data()
              self.dtype = np.float32
              expand_shapes_tensor = []
              for index, ele in enumerate(self.expand_shape):
                  expand_shapes_tensor.append(("x" + str(index), np.ones(
                      (1)).astype('int32') * ele))
  
              self.inputs = {
                  'X': np.random.random(self.ori_shape).astype(self.dtype),
                  'expand_shapes_tensor': expand_shapes_tensor,
              }
              self.attrs = {"shape": self.infer_expand_shape}
              output = np.tile(self.inputs['X'], self.expand_times)
              self.outputs = {'Out': output}
  
          def init_data(self):
              self.ori_shape = [100]
              self.expand_times = [1]
              self.expand_shape = [100]
              self.infer_expand_shape = [-1]
  
      class TestExpandV2OpRank2_Corner_tensor_attr(
              TestExpandV2OpXPURank1_tensor_attr):
  
          def init_data(self):
              self.ori_shape = [12, 14]
              self.expand_times = [1, 1]
              self.expand_shape = [12, 14]
              self.infer_expand_shape = [12, -1]
  
      # Situation 3: shape is a tensor
      class TestExpandV2XPUOp_tensor(TestExpandV2XPUOp):
  
          def setUp(self):
              self.set_xpu()
              self.place = paddle.XPUPlace(0)
              self.op_type = "expand_v2"
              self.init_data()
              self.dtype = np.float32
  
              self.inputs = {
                  'X': np.random.random(self.ori_shape).astype(self.dtype),
                  'Shape': np.array(self.expand_shape).astype("int32"),
              }
              self.attrs = {}
              output = np.tile(self.inputs['X'], self.expand_times)
              self.outputs = {'Out': output}
  
          def init_data(self):
              self.ori_shape = [100]
              self.expand_times = [2, 1]
              self.expand_shape = [2, 100]
  
  
  # Situation 5: input x is int32
  # skip grad check for int32
  class TestExpandV2OpInteger(XPUOpTest):
  
      def init_type(self):
          self.dtype = 'int32'
  
      def setUp(self):
          self.set_xpu()
          self.init_type()
          self.place = paddle.XPUPlace(0)
          self.op_type = "expand_v2"
          self.inputs = {
              'X': np.random.randint(10, size=(2, 4, 20)).astype(self.dtype)
          }
          self.attrs = {'shape': [2, 4, 20]}
          output = np.tile(self.inputs['X'], (1, 1, 1))
          self.outputs = {'Out': output}
  
      def set_xpu(self):
          self.__class__.use_xpu = True
  
      def test_check_output(self):
          self.check_output_with_place(self.place)
  
      def test_check_grad(self):
          pass
  
  
  # Test python API
  class TestExpandV2API(unittest.TestCase):
  
      def test_static(self):
          with fluid.program_guard(fluid.Program(), fluid.Program()):
              input = np.random.random([12, 14]).astype("float32")
              x = fluid.layers.data(name='x',
                                    shape=[12, 14],
                                    append_batch_size=False,
                                    dtype="float32")
  
              positive_2 = fluid.layers.fill_constant([1], "int32", 12)
              expand_shape = fluid.layers.data(name="expand_shape",
                                               shape=[2],
                                               append_batch_size=False,
                                               dtype="int32")
  
              out_1 = paddle.expand(x, shape=[12, 14])
              out_2 = paddle.expand(x, shape=[positive_2, 14])
              out_3 = paddle.expand(x, shape=expand_shape)
  
              g0 = fluid.backward.calc_gradient(out_2, x)
  
              exe = fluid.Executor(place=paddle.XPUPlace(0))
              res_1, res_2, res_3 = exe.run(fluid.default_main_program(),
                                            feed={
                                                "x":
                                                input,
                                                "expand_shape":
                                                np.array([12, 14]).astype("int32")
                                            },
                                            fetch_list=[out_1, out_2, out_3])
  
              assert np.array_equal(res_1, np.tile(input, (1, 1)))
              assert np.array_equal(res_2, np.tile(input, (1, 1)))
              assert np.array_equal(res_3, np.tile(input, (1, 1)))
  
  
  support_types = get_xpu_op_support_types('expand_v2')
  for stype in support_types:
      create_test_class(globals(), XPUTestExpandV2Op, stype)
  
  if __name__ == "__main__":
      unittest.main()
   #   Copyright (c) 2021 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 numpy
  import copy
  
  
  def _get_nested_list_shape(nested_list):
      """
      Get the shape of a nested_list.
      """
      result = []
      while isinstance(nested_list, list):
          result.append(len(nested_list))
          nested_list = nested_list[0]
      return result
  
  
  def _flatten_nested_list(nested_list):
      """
      Get a list of all items in a nested_list.
      Ref: https://stackoverflow.com/questions/952914/how-to-make-a-flat-list-out-of-a-list-of-lists
      """
      result = numpy.array(nested_list).flatten().tolist()
      return result
  
  
  class ProcessMesh(object):
      r"""
      The class `Processmesh` describes the topology of logical processes. 
      A mesh is an N-dimensional array. The shape of the N-dimensional
      array represents the topology of logical processes and every
      element of the N-dimensional array represent a logical process. For
      example, the 2-dimensional array [[2, 4, 5], [0, 1, 3]]
      illustrates six logical processes organized as the topology [2, 3],
      i.e., the shape of the 2-dimensional array. With the above topology,
      there are two parallel groups, where the first parallel group has a
      parallel degree of 2 and the second one has a parallel degree of 3.
      And the first logical process is the one with id=2.
  
      Args:
          mesh (list): an N-dimensional array (nested list) describes the toplogy
              of logical processes. The shape of the N-dimensional array
              represents the topology of logical processes and every 
              element of the N-dimensional array represents a logical process.
      
      Returns:
          None
  
      Raises:
          ValueError: If `mesh` is not an instance of list.
  
      Examples:
          .. code-block:: python
  
              import paddle
              import paddle.distributed as dist
              
              paddle.enable_static()
              
              mesh = dist.ProcessMesh([[2, 4, 5], [0, 1, 3]])
              assert mesh.topology == [2, 3]
              assert mesh.processes == [2, 4, 5, 0, 1, 3]
  
      """
  
      def __init__(self, mesh):
          if mesh is None or not isinstance(mesh, list):
              raise ValueError('mesh must be an instance of list.')
  
          processes = _flatten_nested_list(mesh)
  
          assert all(isinstance(p, int) for p in processes), \
              ("All elements of mesh must be integer")
  
          assert min(processes) >= 0, ('All elements of mesh must be >= 0.')
  
          unique_processes = set(processes)
          assert len(unique_processes) == len(processes), (
              'All elements of mesh must be unique.')
  
          self._topology = _get_nested_list_shape(mesh)
          self._processes = processes
  
          # Store all process meshes
          from .dist_context import get_default_distributed_context
          default_dist_cxt = get_default_distributed_context()
          default_dist_cxt.add_process_mesh(self)
          # Add new processes to process group 0
          from .process_group import get_process_group
          pg0 = get_process_group(0)
          pg0.add_ranks(self.processes)
  
      @property
      def topology(self):
          r"""
          Get the topology of logical processes belonging to this ProcessMesh.
          This is the shape of `mesh` used to initialized this ProcessMesh.
          """
          return self._topology
  
      @property
      def processes(self):
          r"""
          Get a list of all processes belonging to this ProcessMesh.
          """
          return self._processes
  
      @property
      def ndim(self):
          r"""
          Get the number of dimension of ProcessMesh.
          """
          return len(self._topology)
  
      def __eq__(self, other):
          if not isinstance(other, ProcessMesh):
              return False
          if self.topology != other.topology or self.processes != other.processes:
              return False
          return True
  
      def __ne__(self, other):
          return not self.__eq__(other)
  
      def __str__(self):
          str = "shape {} and process group {}".format(self.topology,
                                                       self.processes)
          return str
   # Copyright (c) 2019 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.
  
  from __future__ import print_function
  
  import unittest
  import numpy as np
  
  import paddle.fluid as fluid
  from paddle.fluid.dygraph.nn import Conv2D, Pool2D, Linear
  from paddle.fluid.framework import _test_eager_guard
  
  
  class SimpleImgConvPool(fluid.dygraph.Layer):
  
      def __init__(self,
                   num_channels,
                   num_filters,
                   filter_size,
                   pool_size,
                   pool_stride,
                   pool_padding=0,
                   pool_type='max',
                   global_pooling=False,
                   conv_stride=1,
                   conv_padding=0,
                   conv_dilation=1,
                   conv_groups=1,
                   act=None,
                   use_cudnn=False,
                   dtype='float32',
                   param_attr=None,
                   bias_attr=None):
          super(SimpleImgConvPool, self).__init__()
  
          self._conv2d = Conv2D(num_channels=num_channels,
                                num_filters=num_filters,
                                filter_size=filter_size,
                                stride=conv_stride,
                                padding=conv_padding,
                                dilation=conv_dilation,
                                groups=conv_groups,
                                param_attr=param_attr,
                                bias_attr=bias_attr,
                                use_cudnn=use_cudnn,
                                dtype=dtype,
                                act=act)
  
          self._pool2d = Pool2D(pool_size=pool_size,
                                pool_type=pool_type,
                                pool_stride=pool_stride,
                                pool_padding=pool_padding,
                                global_pooling=global_pooling,
                                use_cudnn=use_cudnn)
  
      def forward(self, inputs):
          x = self._conv2d(inputs)
          x = self._pool2d(x)
          return x
  
  
  class MNIST(fluid.dygraph.Layer):
  
      def __init__(self, dtype="float32"):
          super(MNIST, self).__init__()
  
          self._simple_img_conv_pool_1 = SimpleImgConvPool(num_channels=3,
                                                           num_filters=20,
                                                           filter_size=5,
                                                           pool_size=2,
                                                           pool_stride=2,
                                                           act="relu",
                                                           dtype=dtype,
                                                           use_cudnn=True)
  
          self._simple_img_conv_pool_2 = SimpleImgConvPool(num_channels=20,
                                                           num_filters=50,
                                                           filter_size=5,
                                                           pool_size=2,
                                                           pool_stride=2,
                                                           act="relu",
                                                           dtype=dtype,
                                                           use_cudnn=True)
  
          self.pool_2_shape = 50 * 53 * 53
          SIZE = 10
          scale = (2.0 / (self.pool_2_shape**2 * SIZE))**0.5
          self._linear = Linear(
              self.pool_2_shape,
              10,
              param_attr=fluid.param_attr.ParamAttr(
                  initializer=fluid.initializer.NormalInitializer(loc=0.0,
                                                                  scale=scale)),
              act="softmax",
              dtype=dtype)
  
      def forward(self, inputs, label):
          x = self._simple_img_conv_pool_1(inputs)
          x = self._simple_img_conv_pool_2(x)
          x = fluid.layers.reshape(x, shape=[-1, self.pool_2_shape])
          cost = self._linear(x)
          loss = fluid.layers.cross_entropy(cost, label)
          avg_loss = fluid.layers.mean(loss)
          return avg_loss
  
  
  class TestMnist(unittest.TestCase):
  
      def func_mnist_fp16(self):
          if not fluid.is_compiled_with_cuda():
              return
          x = np.random.randn(1, 3, 224, 224).astype("float16")
          y = np.random.randint(10, size=[1, 1], dtype="int64")
          with fluid.dygraph.guard(fluid.CUDAPlace(0)):
              model = MNIST(dtype="float16")
              x = fluid.dygraph.to_variable(x)
              y = fluid.dygraph.to_variable(y)
              loss = model(x, y)
              print(loss.numpy())
  
      def test_mnist_fp16(self):
          with _test_eager_guard():
              self.func_mnist_fp16()
          self.func_mnist_fp16()
  
  
  if __name__ == "__main__":
      unittest.main()
   #  Copyright (c) 2021 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
  import paddle
  import paddle.static
  from paddle.fluid.tests.unittests.ipu.op_test_ipu import IPUOpTest
  
  
  @unittest.skipIf(not paddle.is_compiled_with_ipu(),
                   "core is not compiled with IPU")
  class TestMul(IPUOpTest):
  
      def setUp(self):
          self.set_atol()
          self.set_training()
          self.set_test_op()
  
      @property
      def fp16_enabled(self):
          if IPUOpTest.use_ipumodel():
              return False
          else:
              return True
  
      def set_test_op(self):
          self.op = paddle.fluid.layers.elementwise_mul
  
      def set_feed_attr(self):
          self.feed_shape = [x.shape for x in self.feed_fp32.values()]
          self.feed_list = list(self.feed_fp32.keys())
  
      @IPUOpTest.static_graph
      def build_model(self):
          x = paddle.static.data(name=self.feed_list[0],
                                 shape=self.feed_shape[0],
                                 dtype='float32')
          y = paddle.static.data(name=self.feed_list[1],
                                 shape=self.feed_shape[1],
                                 dtype='float32')
          out = self.op(x, y, **self.attrs)
          self.fetch_list = [out.name]
  
      def run_model(self, exec_mode):
          self.run_op_test(exec_mode)
  
      def run_test_base(self):
          for m in IPUOpTest.ExecutionMode:
              if not self.skip_mode(m):
                  self.build_model()
                  self.run_model(m)
          self.check()
  
      def test_case0(self):
          data_x = np.random.uniform(size=(2, 3, 4, 5))
          data_y = np.random.uniform(size=(2, 3, 4, 5))
  
          self.feed_fp32 = {
              "x": data_x.astype('float32'),
              "y": data_y.astype('float32'),
          }
          self.feed_fp16 = {
              "x": data_x.astype('float16'),
              "y": data_y.astype('float16'),
          }
          self.attrs = {}
          self.set_feed_attr()
          self.run_test_base()
  
      def test_case1(self):
          data_x = np.random.uniform(size=(2, 3, 4, 5))
          data_y = np.random.uniform(size=(3, 4))
          self.feed_fp32 = {
              "x": data_x.astype('float32'),
              "y": data_y.astype('float32'),
          }
          self.feed_fp16 = {
              "x": data_x.astype('float16'),
              "y": data_y.astype('float16'),
          }
          self.set_feed_attr()
          self.attrs = {"axis": 1}
          self.run_test_base()
  
      def test_case2(self):
          data_x = np.random.uniform(size=(2, 3, 4, 5))
          data_y = np.random.uniform(size=(5))
          self.feed_fp32 = {
              "x": data_x.astype('float32'),
              "y": data_y.astype('float32'),
          }
          self.feed_fp16 = {
              "x": data_x.astype('float16'),
              "y": data_y.astype('float16'),
          }
          self.set_feed_attr()
          self.attrs = {"axis": -1}
          self.run_test_base()
  
      def test_case3(self):
          data_x = np.random.uniform(size=(2, 3, 4, 5))
          data_y = np.random.uniform(size=(2))
          self.feed_fp32 = {
              "x": data_x.astype('float32'),
              "y": data_y.astype('float32'),
          }
          self.feed_fp16 = {
              "x": data_x.astype('float16'),
              "y": data_y.astype('float16'),
          }
          self.set_feed_attr()
          self.attrs = {"axis": 0}
          self.run_test_base()
  
  
  class TestAdd(TestMul):
  
      def set_test_op(self):
          self.op = paddle.fluid.layers.elementwise_add
  
  
  class TestSub(TestMul):
  
      def set_test_op(self):
          self.op = paddle.fluid.layers.elementwise_sub
  
  
  class TestDiv(TestMul):
  
      def set_test_op(self):
          self.op = paddle.fluid.layers.elementwise_div
  
  
  class TestMin(TestMul):
  
      def set_test_op(self):
          self.op = paddle.fluid.layers.elementwise_min
  
  
  class TestMax(TestMul):
  
      def set_test_op(self):
          self.op = paddle.fluid.layers.elementwise_max
  
  
  class TestPow(TestMul):
  
      def set_test_op(self):
          self.op = paddle.fluid.layers.elementwise_pow
  
  
  class TestMod(TestMul):
  
      def set_atol(self):
          self.atol = 1e-7
          self.rtol = 1e-5
          self.atol_fp16 = 1e-2
          self.rtol_fp16 = 1e-3
  
      def set_test_op(self):
          self.op = paddle.fluid.layers.elementwise_mod
  
  
  if __name__ == "__main__":
      unittest.main()
  
apis: numpy.tile numpy.ones numpy.random.seed numpy.random.random numpy.array numpy.random.randint
apis: numpy.array
apis: numpy.random.randint numpy.random.randn
apis: numpy.random.uniform