tools/shared/cwrap_common.py

# this code should be common among cwrap and ATen preprocessing
# for now, I have put it in one place but right now is copied out of cwrap

import copy
from typing import Any, Dict, Iterable, List, Union

Arg = Dict[str, Any]

def parse_arguments(args: List[Union[str, Arg]]) -> List[Arg]:
    new_args = []
    for arg in args:
        # Simple arg declaration of form "<type> <name>"
        if isinstance(arg, str):
            t, _, name = arg.partition(' ')
            new_args.append({'type': t, 'name': name})
        elif isinstance(arg, dict):
            if 'arg' in arg:
                arg['type'], _, arg['name'] = arg['arg'].partition(' ')
                del arg['arg']
            new_args.append(arg)
        else:
            raise AssertionError()
    return new_args


Declaration = Dict[str, Any]


def set_declaration_defaults(declaration: Declaration) -> None:
    if 'schema_string' not in declaration:
        # This happens for legacy TH bindings like
        # _thnn_conv_depthwise2d_backward
        declaration['schema_string'] = ''
    declaration.setdefault('arguments', [])
    declaration.setdefault('return', 'void')
    if 'cname' not in declaration:
        declaration['cname'] = declaration['name']
    if 'backends' not in declaration:
        declaration['backends'] = ['CPU', 'CUDA']
    assert 'api_name' not in declaration
    declaration['api_name'] = declaration['name']
    # NB: keep this in sync with gen_autograd.py
    if declaration.get('overload_name'):
        declaration['type_wrapper_name'] = "{}_{}".format(
            declaration['name'], declaration['overload_name'])
    else:
        declaration['type_wrapper_name'] = declaration['name']
    # TODO: Uggggh, parsing the schema string here, really???
    declaration['operator_name_with_overload'] = declaration['schema_string'].split('(')[0]
    if declaration['schema_string']:
        declaration['unqual_schema_string'] = declaration['schema_string'].split('::')[1]
        declaration['unqual_operator_name_with_overload'] = declaration['operator_name_with_overload'].split('::')[1]
    else:
        declaration['unqual_schema_string'] = ''
        declaration['unqual_operator_name_with_overload'] = ''
    # Simulate multiple dispatch, even if it's not necessary
    if 'options' not in declaration:
        declaration['options'] = [{
            'arguments': copy.deepcopy(declaration['arguments']),
            'schema_order_arguments': copy.deepcopy(declaration['schema_order_arguments']),
        }]
        del declaration['arguments']
        del declaration['schema_order_arguments']
    # Parse arguments (some of them can be strings)
    for option in declaration['options']:
        option['arguments'] = parse_arguments(option['arguments'])
        option['schema_order_arguments'] = parse_arguments(option['schema_order_arguments'])
    # Propagate defaults from declaration to options
    for option in declaration['options']:
        for k, v in declaration.items():
            # TODO(zach): why does cwrap not propagate 'name'? I need it
            # propagaged for ATen
            if k != 'options':
                option.setdefault(k, v)

# TODO(zach): added option to remove keyword handling for C++ which cannot
# support it.

Option = Dict[str, Any]


def filter_unique_options(
    options: Iterable[Option],
    allow_kwarg: bool,
    type_to_signature: Dict[str, str],
    remove_self: bool,
) -> List[Option]:
    def exclude_arg(arg: Arg) -> bool:
        return arg['type'] == 'CONSTANT'  # type: ignore[no-any-return]

    def exclude_arg_with_self_check(arg: Arg) -> bool:
        return exclude_arg(arg) or (remove_self and arg['name'] == 'self')

    def signature(option: Option, num_kwarg_only: int) -> str:
        if num_kwarg_only == 0:
            kwarg_only_count = None
        else:
            kwarg_only_count = -num_kwarg_only
        arg_signature = '#'.join(
            type_to_signature.get(arg['type'], arg['type'])
            for arg in option['arguments'][:kwarg_only_count]
            if not exclude_arg_with_self_check(arg))
        if kwarg_only_count is None:
            return arg_signature
        kwarg_only_signature = '#'.join(
            arg['name'] + '#' + arg['type']
            for arg in option['arguments'][kwarg_only_count:]
            if not exclude_arg(arg))
        return arg_signature + "#-#" + kwarg_only_signature
    seen_signatures = set()
    unique = []
    for option in options:
        # if only check num_kwarg_only == 0 if allow_kwarg == False
        limit = len(option['arguments']) if allow_kwarg else 0
        for num_kwarg_only in range(0, limit + 1):
            sig = signature(option, num_kwarg_only)
            if sig not in seen_signatures:
                if num_kwarg_only > 0:
                    for arg in option['arguments'][-num_kwarg_only:]:
                        arg['kwarg_only'] = True
                unique.append(option)
                seen_signatures.add(sig)
                break
    return unique


def sort_by_number_of_args(declaration: Declaration, reverse: bool = True) -> None:
    def num_args(option: Option) -> int:
        return len(option['arguments'])
    declaration['options'].sort(key=num_args, reverse=reverse)


class Function(object):

    def __init__(self, name: str) -> None:
        self.name = name
        self.arguments: List['Argument'] = []

    def add_argument(self, arg: 'Argument') -> None:
        assert isinstance(arg, Argument)
        self.arguments.append(arg)

    def __repr__(self) -> str:
        return self.name + '(' + ', '.join(a.__repr__() for a in self.arguments) + ')'


class Argument(object):

    def __init__(self, _type: str, name: str, is_optional: bool):
        self.type = _type
        self.name = name
        self.is_optional = is_optional

    def __repr__(self) -> str:
        return self.type + ' ' + self.name


def parse_header(path: str) -> List[Function]:
    with open(path, 'r') as f:
        lines: Iterable[Any] = f.read().split('\n')

    # Remove empty lines and prebackend directives
    lines = filter(lambda l: l and not l.startswith('#'), lines)
    # Remove line comments
    lines = (l.partition('//') for l in lines)
    # Select line and comment part
    lines = ((l[0].strip(), l[2].strip()) for l in lines)
    # Remove trailing special signs
    lines = ((l[0].rstrip(');').rstrip(','), l[1]) for l in lines)
    # Split arguments
    lines = ((l[0].split(','), l[1]) for l in lines)
    # Flatten lines
    new_lines = []
    for l, c in lines:
        for split in l:
            new_lines.append((split, c))
    lines = new_lines
    del new_lines
    # Remove unnecessary whitespace
    lines = ((l[0].strip(), l[1]) for l in lines)
    # Remove empty lines
    lines = filter(lambda l: l[0], lines)
    generic_functions = []
    for l, c in lines:
        if l.startswith('TH_API void THNN_'):
            fn_name = l[len('TH_API void THNN_'):]
            if fn_name[0] == '(' and fn_name[-2] == ')':
                fn_name = fn_name[1:-2]
            else:
                fn_name = fn_name[:-1]
            generic_functions.append(Function(fn_name))
        elif l.startswith('TORCH_CUDA_CPP_API void THNN_'):
            fn_name = l[len('TORCH_CUDA_CPP_API void THNN_'):]
            if fn_name[0] == '(' and fn_name[-2] == ')':
                fn_name = fn_name[1:-2]
            else:
                fn_name = fn_name[:-1]
            generic_functions.append(Function(fn_name))
        elif l.startswith('TORCH_CUDA_CU_API void THNN_'):
            fn_name = l[len('TORCH_CUDA_CU_API void THNN_'):]
            if fn_name[0] == '(' and fn_name[-2] == ')':
                fn_name = fn_name[1:-2]
            else:
                fn_name = fn_name[:-1]
            generic_functions.append(Function(fn_name))
        elif l:
            t, name = l.split()
            if '*' in name:
                t = t + '*'
                name = name[1:]
            generic_functions[-1].add_argument(
                Argument(t, name, '[OPTIONAL]' in c))
    return generic_functions