cupy_numpy_comparison.py

import inspect
from inspect import Signature, Parameter
from typing import Tuple, List, Set, Dict
from types import ModuleType
from html import escape

import numpy as np
import scipy as sc
import cupy as cp

DiffResult = Tuple[str, str, str, bool, bool]


def lcs(left: Signature, right: Signature,
        left_module: ModuleType,
        right_module: ModuleType) -> List[DiffResult]:
    params_l = left.parameters
    params_r = right.parameters
    param_names_l = list(params_l)
    param_names_r = list(params_r)
    m = len(param_names_l)
    n = len(param_names_r)
    backtrack = np.zeros((m + 1, n + 1))
    for i in range(0, m):
        for j in range(0, n):
            if param_names_l[i] == param_names_r[j]:
                backtrack[i + 1, j + 1] = backtrack[i, j] + 1
            else:
                backtrack[i + 1, j + 1] = max(
                    backtrack[i + 1, j], backtrack[i, j])
    result_stack = []
    queue = [(m, n)]
    while queue != []:
        i, j = queue.pop(0)
        if i > 0 and j > 0 and param_names_l[i - 1] == param_names_r[j - 1]:
            param_left_name = param_names_l[i - 1]
            param_right_name = param_names_r[j - 1]
            param_l = params_l[param_left_name]
            param_r = params_r[param_right_name]
            if param_l.default is inspect._empty:
                if param_r.default is inspect._empty:
                    result_stack.append(
                        (param_left_name, 'same position', 'both non-kwarg',
                         True, True))
                else:
                    result_stack.append(
                        (param_left_name, 'same position',
                         f'{left_module.__name__} is non-kwarg, '
                         f'{right_module.__name__} is kwarg', True, False))
            else:
                if param_r.default is not inspect._empty:
                    result_stack.append(
                        (param_left_name, 'same position', 'both kwarg',
                        True, True))
                else:
                    result_stack.append(
                        (param_left_name, 'same position',
                         f'{left_module.__name__} is kwarg, '
                         f'{right_module.__name__} is non-kwarg', True, False))
            queue.append((i - 1, j - 1))
        elif j > 0 and (i == 0 or backtrack[i, j - 1] >= backtrack[i - 1, j]):
            param_right_name = param_names_r[j - 1]
            param_r = params_r[param_right_name]
            extra = 'is kwarg'
            if param_r.default is inspect._empty:
                extra = 'is non-kwarg'
            result_stack.append(
                (param_right_name,
                 f'extra arg on {right_module.__name__}', extra,
                 False, False))
            queue.append((i, j - 1))
        elif i > 0 and (j == 0 or backtrack[i, j - 1] < backtrack[i - 1, j]):
            param_left_name = param_names_l[i - 1]
            param_l = params_l[param_left_name]
            extra = 'is kwarg'
            if param_l.default is Parameter.default:
                extra = 'is non-kwarg'
            result_stack.append(
                (param_left_name,
                 f'extra arg on {left_module.__name__}', extra,
                 False, False))
            queue.append((i - 1, j))
    return list(reversed(result_stack))


numpy_dir = dir(np)
cupy_dir = dir(cp)

checks = [
    ('callable', callable),
    ('function', inspect.isfunction),
    ('function', inspect.ismethod),
    ('module', inspect.ismodule),
    # 'class': inspect.isclass
]

manual_check = []

result: Dict[str, Dict[str, List[DiffResult]]] = {'main': {}}
main_results = result['main']
cp_np_shared_funcs = set(numpy_dir) & set(cupy_dir)
stack: List[Tuple[
    Set[str], ModuleType, ModuleType, Dict[str, List[DiffResult]]]] = [
        (cp_np_shared_funcs, cp, np, main_results)]

while stack != []:
    members, mod1, mod2, res = stack.pop(0)
    for mem_name in members:
        if mem_name.startswith('_'):
            continue
        mem1 = getattr(mod1, mem_name)
        mem2 = getattr(mod2, mem_name)
        check1 = {chk for chk, chk_fn in checks if chk_fn(mem1)}
        check2 = {chk for chk, chk_fn in checks if chk_fn(mem2)}
        if len(check1) > 0 and len(check2) > 0:
            print(f'Comparing {mem_name}')
            if 'function' in check1:
                if 'callable' in check2:
                    if 'function' not in check2:
                        manual_check.append(f'{mod1.__name__}.{mem_name}')
                    else:
                        sig1 = inspect.signature(mem1)
                        sig2 = inspect.signature(mem2)
                        diff = lcs(sig1, sig2, mod1, mod2)
                        all_match = all([
                            position_match and kwarg_match
                            for _, _, _, position_match, kwarg_match in diff])
                        if not all_match:
                            res[mem_name] = diff
            elif 'callable' in check1:
                if {'function', 'callable'} - check2 == set({}):
                    manual_check.append(f'{mod2.__name__}.{mem_name}')
            elif 'module' in check1:
                if 'module' in check2:
                    mod_res = {}
                    result[mem_name] = mod_res
                    lmod = getattr(mod1, mem_name)
                    rmod = getattr(mod2, mem_name)
                    dir_lmod = dir(lmod)
                    dir_rmod = dir(rmod)
                    shared_funcs = set(dir_lmod) & set(dir_rmod)
                    stack.append((shared_funcs, lmod, rmod, mod_res))


result = {k: result[k] for k in result if len(result[k]) > 0}
table_header = '| Function | Observations |\n|:----:|:-----:|'
text_results = []

for namespace in result:
    namespace_results = []
    namespace_results.append(f'## {namespace} namespace\n')
    namespace_results.append(table_header)
    namespace_arg_results = result[namespace]
    for func_name in sorted(namespace_arg_results):
        func_args = namespace_arg_results[func_name]
        wrong_args = []
        for arg in func_args:
            (arg_name, position_obs, req_const_obs,
             correct_pos, same_req_const) = arg
            if not (correct_pos and same_req_const):
                wrong_args.append(
                    f'`{arg_name}`: {position_obs}, {req_const_obs}')
        observations = '<br/>'.join(wrong_args)
        namespace_results.append(f'| `{func_name}` | {observations} |')
    text_results.append('\n'.join(namespace_results))

print('\n\n'.join(text_results))