op_test.py

import unittest
import numpy as np
import itertools
import paddle.v2.framework.core as core
from paddle.v2.framework.op import Operator


def grad_var_name(var_name):
    return var_name + "@GRAD"


def remove_grad_var_name(var_name):
    return var_name[0:-5]


def create_op(scope, op_type, inputs, outputs, attrs=None):
    kwargs = dict()

    for ins in Operator.get_op_inputs(op_type):
        in_name = ins[0]
        in_dup = ins[1]
        if in_name in inputs:
            kwargs[in_name] = []
            if in_dup:
                sub_in = inputs[in_name]
                for sub_in_name in sub_in:
                    var = scope.new_var(sub_in_name)
                    tensor = var.get_tensor()
                    kwargs[in_name].append(sub_in_name)
            else:
                var = scope.new_var(in_name)
                tensor = var.get_tensor()
                kwargs[in_name].append(in_name)

    for outs in Operator.get_op_outputs(op_type):
        out_name = outs[0]
        out_dup = outs[1]
        if out_name in outputs:
            kwargs[out_name] = []
            if out_dup:
                sub_in = outputs[out_name]
                for sun_in_name in sub_in:
                    var = scope.new_var(sun_in_name)
                    tensor = var.get_tensor()
                    kwargs[out_name].append(sun_in_name)
            else:
                var = scope.new_var(out_name)
                tensor = var.get_tensor()
                kwargs[out_name].append(out_name)

    # for attr_name in Operator.get_op_attr_names(op_type):
    #	  kwargs[attr_name] = attrs[attr_name]
    return Operator(op_type, **kwargs)


def set_input(scope, op, inputs, place):
    for ins in Operator.get_op_inputs(op.type()):
        in_name = ins[0]
        in_dup = ins[1]
        if in_name in inputs:
            if in_dup:
                sub_in = inputs[in_name]
                for sub_in_name in sub_in:
                    var = scope.find_var(sub_in_name)
                    tensor = var.get_tensor()
                    arr = sub_in[sub_in_name]
                    tensor.set_dims(arr.shape)
                    tensor.set(arr, place)
            else:
                var = scope.find_var(in_name)
                tensor = var.get_tensor()
                arr = inputs[in_name]
                tensor.set_dims(arr.shape)
                tensor.set(arr, place)


def set_output_grad(scope, op, outputs, place):
    for outs in Operator.get_op_outputs(op.type()):
        out_name = outs[0]
        out_dup = outs[1]
        if out_name in outputs:
            if out_dup:
                sub_out = outputs[out_name]
                for sub_out_name in sub_out:
                    out_tensor = scope.find_var(sub_out_name).get_tensor()
                    grad_tensor = scope.new_var(grad_var_name(
                        sub_out_name)).get_tensor()
                    grad_tensor.set_dims(out_tensor.shape())
                    data = np.ones(out_tensor.shape(), dtype=np.float32)
                    grad_tensor.set(data, place)
            else:
                out_tensor = scope.find_var(out_name).get_tensor()
                grad_tensor = scope.new_var(grad_var_name(out_name)).get_tensor(
                )
                grad_tensor.set_dims(out_tensor.shape())
                data = np.ones(out_tensor.shape(), dtype=np.float32)
                grad_tensor.set(data, place)


def get_numeric_gradient(scope,
                         op,
                         inputs,
                         input_to_check,
                         output_name,
                         delta=0.005,
                         in_place=False):

    set_input(scope, op, inputs, core.CPUPlace())
    op.infer_shape(scope)

    tensor_to_check = scope.find_var(input_to_check).get_tensor()

    def product(dim):
        return reduce(lambda a, b: a * b, dim, 1)

    ctx = core.DeviceContext.create(core.CPUPlace())

    def get_output():
        op.run(scope, ctx)
        return np.array(scope.find_var(output_name).get_tensor()).sum()

    tensor_to_check = scope.find_var(input_to_check).get_tensor()
    tensor_size = product(tensor_to_check.get_dims())
    gradient_flat = np.zeros(shape=(tensor_size, ), dtype='float32')
    # we only compute gradient of one element each time.
    # we use a for loop to compute the gradient of every element.
    for i in xrange(tensor_size):
        if in_place:
            set_input(op, inputs, core.CPUPlace())

        # get one input element throw it's index i.
        origin = tensor_to_check.get_float_element(i)
        # add delta to it, run op and then get the sum of the result tensor.
        x_pos = origin + delta
        tensor_to_check.set_float_element(i, x_pos)
        y_pos = get_output()

        if in_place:
            set_input(op, inputs, core.CPUPlace())

        x_neg = origin - delta
        tensor_to_check.set_float_element(i, x_neg)
        y_neg = get_output()

        tensor_to_check.set_float_element(i, origin)
        gradient_flat[i] = (y_pos - y_neg) / delta / 2

    return gradient_flat.reshape(tensor_to_check.get_dims())


def get_backward_op(scope, op, no_grad_set):
    backward_op = core.Operator.backward(op, no_grad_set)
    for input in backward_op.inputs_names():
        var = scope.new_var(input)
        var.get_tensor()
    for output in backward_op.outputs_names():
        var = scope.new_var(output)
        var.get_tensor()
    return backward_op


def get_gradient(scope, op, inputs, outputs, grad_name, place,
                 no_grad_set=None):
    ctx = core.DeviceContext.create(place)

    set_input(scope, op, inputs, place)

    op.infer_shape(scope)
    op.run(scope, ctx)

    if no_grad_set is None:
        no_grad_set = set()

    backward_op = get_backward_op(scope, op, no_grad_set)
    set_output_grad(scope, op, outputs, place)

    backward_op.infer_shape(scope)
    backward_op.run(scope, ctx)

    out = np.array(scope.find_var(grad_name).get_tensor())
    return out


class OpTest(unittest.TestCase):
    def check_output(self, place):
        self.scope = core.Scope()
        self.op = create_op(self.scope, self.op_type, self.inputs, self.outputs)
        if isinstance(place, core.GPUPlace) and not self.op.support_gpu():
            return
        set_input(self.scope, self.op, self.inputs, place)
        self.op.infer_shape(self.scope)
        ctx = core.DeviceContext.create(place)
        self.op.run(self.scope, ctx)

        for outs in Operator.get_op_outputs(self.op.type()):
            out_name = outs[0]
            out_dup = outs[1]
            if out_dup:
                sub_out = self.outputs[out_name]
                for sub_out_name in sub_out:
                    actual = np.array(
                        self.scope.find_var(sub_out_name).get_tensor())
                    expect = sub_out[sub_out_name]
                    self.assertTrue(
                        np.allclose(
                            actual, expect, atol=1e-05),
                        "output name: " + out_name + "has diff")
            else:
                actual = np.array(self.scope.find_var(out_name).get_tensor())
                expect = self.outputs[out_name]
                self.assertTrue(
                    np.allclose(
                        actual, expect, atol=1e-05),
                    "output name: " + out_name + "has diff")

    def __assert_is_close(self, numeric_grads, analytic_grads, names,
                          max_relative_error, msg_prefix):

        for a, b, name in itertools.izip(numeric_grads, analytic_grads, names):
            abs_a = np.abs(a)
            abs_a[abs_a < 1e-3] = 1

            diff_mat = np.abs(a - b) / abs_a
            max_diff = np.max(diff_mat)

            def err_msg():
                offset = np.argmax(diff_mat > max_relative_error)
                return "%s Variable %s max gradient diff %f over limit %f, the first " \
                  "error element is %d" % (
                   msg_prefix, name, max_diff, max_relative_error, offset)

            self.assertLessEqual(max_diff, max_relative_error, err_msg())

    def check_grad(self,
                   inputs_to_check,
                   output_name,
                   no_grad_set=None,
                   in_place=False,
                   max_relative_error=0.005):
        self.scope = core.Scope()
        self.op = create_op(self.scope, self.op_type, self.inputs, self.outputs)
        if no_grad_set is None:
            no_grad_set = set()

        numeric_grads = [
            get_numeric_gradient(
                self.scope,
                self.op,
                self.inputs,
                input_to_check,
                output_name,
                in_place=in_place) for input_to_check in inputs_to_check
        ]
        grad_names = [
            grad_var_name(input_to_check) for input_to_check in inputs_to_check
        ]

        places = [core.CPUPlace()]
        if core.is_compile_gpu() and self.op.support_gpu():
            places.append(core.GPUPlace(0))

        for place in places:
            analytic_grads = [
                get_gradient(self.scope, self.op, self.inputs, self.outputs,
                             grad_name, place, no_grad_set)
                for grad_name in grad_names
            ]

            self.__assert_is_close(numeric_grads, analytic_grads, grad_names,
                                   max_relative_error,
                                   "Gradient Check On %s" % str(place))