accuracy_checker.py

#####################################################################################
# The MIT License (MIT)
#
# Copyright (c) 2015-2022 Advanced Micro Devices, Inc. All rights reserved.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#####################################################################################
import argparse
import numpy as np
import migraphx
import onnxruntime as ort
import sys


def parse_args():
    parser = argparse.ArgumentParser(
        description=
        'MIGraphX accuracy checker. Use to verify onnx files to ensure MIGraphX\'s output \
                                                  is within tolerance of onnx runtime\'s expected output.'
    )
    file_args = parser.add_argument_group(title='file type arguments')
    file_args.add_argument('--onnx', type=str, help='path to onnx file')
    file_args.add_argument('--tf', type=str, help='path to tf pb file')
    parser.add_argument('--provider',
                        type=str,
                        default='CPUExecutionProvider',
                        help='execution provider for onnx runtime \
                                (default = CPUExecutionProvider)')
    parser.add_argument('--batch',
                        type=int,
                        default=1,
                        help='batch size (if specified in onnx file)')
    parser.add_argument('--fill1',
                        action='store_true',
                        help='fill all arguments with a value of 1')
    parser.add_argument('--fill0',
                        action='store_true',
                        help='fill all arguments with a value of 0')
    parser.add_argument('--verbose',
                        action='store_true',
                        help='show verbose information (for debugging)')
    parser.add_argument('--tolerance',
                        type=float,
                        default=1e-3,
                        help='accuracy tolerance (default = 1e-3)')
    parser.add_argument('--input-dim',
                        type=str,
                        action='append',
                        help='specify input parameter dimension \
                                with the following format --input_dim input_name:dim0,dim1,dim2...'
                        )
    args = parser.parse_args()

    return args


# taken from ../test_runner.py
def check_correctness(gold_outputs,
                      outputs,
                      rtol=1e-3,
                      atol=1e-3,
                      verbose=False):
    if len(gold_outputs) != len(outputs):
        print('Number of outputs {} is not equal to expected number {}'.format(
            len(outputs), len(gold_outputs)))
        return False

    out_num = len(gold_outputs)
    ret = True
    for i in range(out_num):
        if not np.allclose(gold_outputs[i], outputs[i], rtol, atol):
            ret = False
            if verbose:
                print('\nOutput {} is incorrect ...'.format(i))
                print('Expected value: \n{}'.format(gold_outputs[i]))
                print('......')
                print('Actual value: \n{}\n'.format(outputs[i]))
            else:
                print('Outputs do not match')
                break

    return ret


def get_np_datatype(in_type):
    datatypes = {
        'double_type': np.float64,
        'float_type': np.float32,
        'half_type': np.half,
        'int64_type': np.int64,
        'uint64_type': np.uint64,
        'int32_type': np.int32,
        'uint32_type': np.uint32,
        'int16_type': np.int16,
        'uint16_type': np.uint16,
        'int8_type': np.int8,
        'uint8_type': np.uint8,
        'bool_type': np.bool_
    }
    return datatypes[in_type]


def main():
    args = parse_args()

    use_onnx = True
    if args.onnx == None:
        use_onnx = False
    if not use_onnx and args.tf == None:
        print('Error: please specify either an onnx or tf pb file')
        sys.exit(-1)

    model_name = args.onnx

    batch = args.batch

    custom_inputs = args.input_dim

    input_dims = {}
    if custom_inputs != None:
        for input in custom_inputs:
            input_dim = ''.join(input.split(':')[:-1])
            dims = [int(dim) for dim in input.split(':')[-1].split(',')]
            input_dims[input_dim] = dims

    if use_onnx:
        if not input_dims:
            model = migraphx.parse_onnx(model_name, default_dim_value=batch)
        else:
            model = migraphx.parse_onnx(model_name,
                                        default_dim_value=batch,
                                        map_input_dims=input_dims)
    else:
        model_name = args.tf

        if not input_dims:
            model = migraphx.parse_tf(model_name, batch_size=batch)
        else:
            model = migraphx.parse_tf(model_name,
                                      batch_size=batch,
                                      map_input_dims=input_dims)

    if args.verbose:
        print(model)

    model.compile(migraphx.get_target('gpu'))

    params = {}
    test_inputs = {}
    for name, shape in model.get_parameter_shapes().items():
        if args.verbose:
            print(f'Parameter {name} -> {shape}')
        in_shape = shape.lens()
        in_type = shape.type_string()
        if not args.fill1 and not args.fill0:
            test_input = np.random.rand(*(in_shape)).astype(
                get_np_datatype(in_type))
        elif not args.fill0:
            test_input = np.ones(in_shape).astype(get_np_datatype(in_type))
        else:
            test_input = np.zeros(in_shape).astype(get_np_datatype(in_type))
        test_inputs[name] = test_input
        params[name] = migraphx.argument(test_input)

    pred_migx = np.array(model.run(params)[-1])

    if use_onnx:
        sess = ort.InferenceSession(model_name, providers=[args.provider])

        ort_params = {}
        for input in sess.get_inputs():
            ort_params[input.name] = test_inputs[input.name]

        try:
            pred_fw = sess.run(None, ort_params)[-1]
        except Exception as e:
            if any(input_dims):
                print(
                    'Error: custom input dim may not be compatible with onnx runtime'
                )
            raise e
    else:
        import tensorflow as tf

        def load_tf_graph(model_name):
            with tf.io.gfile.GFile(model_name, 'rb') as f:
                graph_def = tf.compat.v1.GraphDef()
                graph_def.ParseFromString(f.read())

            with tf.compat.v1.Graph().as_default() as graph:
                tf.graph_util.import_graph_def(graph_def)
            return graph

        graph = load_tf_graph(model_name)
        is_nhwc = False
        graph_ops = []
        for op in graph.get_operations():
            graph_ops.append(op.name)
            if 'Conv' in op.node_def.op:
                if 'NHWC' in op.get_attr('data_format').decode('utf-8'):
                    is_nhwc = True
        graph_ops_set = set(graph_ops)
        tf_dict = {}

        for name in test_inputs.keys():
            # graph.get_operations() adds 'import/' to the op name
            tf_name = f'import/{name}'
            if tf_name not in graph_ops_set:
                continue
            x = graph.get_tensor_by_name(f'{tf_name}:0')
            tf_input = test_inputs[name]
            # transpose input for NHWC model
            if tf_input.ndim == 4 and is_nhwc:
                tf_dict[x] = np.transpose(tf_input, (0, 2, 3, 1))
            else:
                tf_dict[x] = tf_input

        # assume last node in graph is output
        # TODO: let user specify op name for output
        y = graph.get_tensor_by_name(f'{graph_ops[-1]}:0')

        with tf.compat.v1.Session(graph=graph) as sess:
            y_out = sess.run(y, feed_dict=tf_dict)
            pred_fw = y_out

    is_correct = check_correctness(pred_fw, pred_migx, args.tolerance,
                                   args.tolerance, args.verbose)
    verbose_string = ' Rerun with --verbose for detailed information.' \
            if not args.verbose else ''
    if is_correct:
        print('PASSED: MIGraphX meets tolerance')
    else:
        print('FAILED: MIGraphX is not within tolerance.' + verbose_string)


if __name__ == '__main__':
    main()