genapi.py

import argparse
import os
import re
import subprocess
import sys


MATCH_API_EXPANSION = re.compile(
    r"__GENERATE_API \$ __GEN_ID_(?P<id>\w+) \$ (?P<decl>.*?) \$",
    re.DOTALL | re.MULTILINE,
)

MATCH_ISR_EXPANSION = re.compile(
    r"__GENERATE_API_ISR \$ __GEN_ID_(?P<id>\w+) \$ (?P<isr>.*?) \$",
    re.DOTALL | re.MULTILINE,
)

MATCH_DECLARATION = re.compile(r"^(?P<typename>.*?)\s*\((?P<args>.*)\)$", re.DOTALL)

MATCH_TYPENAME = re.compile(
    r"^(?P<type>(?:const )?(?:struct |enum |union )?\w+[*\s]+)(?P<name>\w+)$"
)


def sizeof(args):
    """Return a string that describes the size of a list of arguments."""
    return " + ".join(a["sizeof"] for a in args) if args != [] else "0"


def bailout(message, *args, **kwargs):
    fmt = "\x1B[31;1mError\x1B[0m: {}"
    print(fmt.format(message.format(*args, **kwargs)), file=sys.stderr)
    sys.exit(1)


def parse_declarations(source):
    """Parse all declarations in the given source."""
    declarations = []
    for exp in MATCH_API_EXPANSION.finditer(source):
        id = exp.group("id")
        decl = MATCH_DECLARATION.match(exp.group("decl"))

        if decl is None:
            bailout("Error in declaration '{}'", exp.group("decl"))

        typename = MATCH_TYPENAME.match(decl.group("typename"))
        args = []
        args_str = decl.group("args")

        if typename is None:
            bailout("Error in declaration '{}'", exp.group("decl"))

        # Parse arguments
        for arg_str in map(str.strip, args_str.split(",")):
            if arg_str in ["void", ""]:
                continue

            arg = MATCH_TYPENAME.match(arg_str.strip())

            if arg is None:
                bailout("Failed to parse argument '{}'", arg_str.strip())

            args.append(
                {
                    "type": arg.group("type").strip(),
                    "name": arg.group("name"),
                    "sizeof": "sizeof({})".format(arg.group("type").strip()),
                    "offset": sizeof(args),
                }
            )

        declarations.append(
            {
                "id": id,
                "return_type": typename.group("type").strip(),
                "name": typename.group("name"),
                "args": args,
                "args_str": args_str,
            }
        )

    return declarations


def parse_interrupts(source):
    """Parse all isr declarations in the given source."""
    interrupts = []
    for exp in MATCH_ISR_EXPANSION.finditer(source):
        id = exp.group("id")
        isr = exp.group("isr")

        interrupts.append({"id": id, "isr": isr})

    return interrupts


def main():
    parser = argparse.ArgumentParser(
        description="Generate the API stubs from a header file."
    )
    parser.add_argument(
        "-H", "--header", required=True, help="The header to base the definitions on."
    )
    parser.add_argument(
        "-c", "--client", required=True, help="The output client-side c source file."
    )
    parser.add_argument(
        "-s", "--server", required=True, help="The output server-side c source file."
    )
    args = parser.parse_args()

    # Run the preprocessor on the header file to get the API definitions.
    #
    # For this, we first need a source to include the header which contains
    # an alternative definition of the `API` macro that marks definitions in
    # a way we can find later on.
    api_src = """\
#define API(id, def) __GENERATE_API $ __GEN_ID_##id $ def $
#define API_ISR(id, isr) __GENERATE_API_ISR $ __GEN_ID_##id $ isr $
#include "{header}"
""".format(
        header=os.path.relpath(args.header)
    )

    # Evaluate the preprocessor
    source = subprocess.check_output(
        ["gcc", "-E", "-P", "-"], input=api_src.encode()
    ).decode()

    declarations = parse_declarations(source)
    interrupts = parse_interrupts(source)

    fmt_header = {"header": os.path.basename(args.header)}

    # Generate Client {{{
    with open(args.client, "w") as f_client:
        tmp = """\
#include <stdio.h>

#define API_ISR(id, isr)
#include "{header}"
#include "api/caller.h"
"""
        f_client.write(tmp.format(**fmt_header))

        for decl in declarations:
            decl["total_size"] = sizeof(decl["args"])
            tmp = """\

/* Autogenerated stub for {id} */
{return_type} {name}({args_str})
{{
        const int epc__apistub_size = {total_size};
        void*epc__apistub_buffer;

        epc__apistub_buffer = _api_call_start({id}, epc__apistub_size);
        /* TODO: Check if epc__apistub_buffer is not NULL */

"""
            f_client.write(tmp.format(**decl))

            for i, arg in enumerate(decl["args"]):
                tmp = """\
        *({type}*)(epc__apistub_buffer + {offset}) = {name};
"""
                f_client.write(tmp.format(**arg))

            if decl["return_type"] == "void":
                # Don't return if return type is void
                tmp = """\

        _api_call_transact(epc__apistub_buffer);
}}
"""
                f_client.write(tmp.format(**decl))
            else:
                tmp = """\

        return *({return_type}*)_api_call_transact(epc__apistub_buffer);
}}
"""
                f_client.write(tmp.format(**decl))

        tmp = """\


/* Weakly linked stubs for ISRs */
"""
        f_client.write(tmp)

        for isr in interrupts:
            tmp = """\
void {isr}(api_int_id_t id)
        __attribute__((weak, alias("__epic_isr_default_handler")));
"""
            f_client.write(tmp.format(**isr))

        tmp = """\

/* Default handler stub */
__attribute__((weak)) void epic_isr_default_handler(api_int_id_t id)
{
        ;
}

/*
 * This function is needed because aliasing the weak default handler will
 * lead to issues.
 */
void __epic_isr_default_handler(api_int_id_t id)
{
        epic_isr_default_handler(id);
}

/*
 * __dispatch_isr() will be called from the actual isr which was triggered
 * by core 0.  It will then call the appropriate isr.
 */
void __dispatch_isr(api_int_id_t id)
{
        switch (id) {
"""
        f_client.write(tmp)

        for isr in interrupts:
            tmp = """\
        case {id}:
                {isr}(id);
                break;
"""
            f_client.write(tmp.format(**isr))

        tmp = """\
        case (-1):
                /* Ignore a spurious interrupt */
                break;
        default:
                epic_isr_default_handler(id);
                break;
        }
}
"""
        f_client.write(tmp)
    # END: Generate Client }}}

    # Generate Dispatcher {{{
    with open(args.server, "w") as f_dispatcher:
        tmp = """\
#include "modules/log.h"
#include "{header}"

void __api_dispatch_call(uint32_t id, void*epc__apistub_buffer)
{{
        switch (id) {{
"""
        f_dispatcher.write(tmp.format(**fmt_header))

        for decl in declarations:
            if decl["return_type"] == "void":
                tmp = """\
        case {id}:
                {name}("""
                f_dispatcher.write(tmp.format(**decl))
            else:
                tmp = """\
        case {id}:
                *(({return_type}*)epc__apistub_buffer) = {name}("""
                f_dispatcher.write(tmp.format(**decl))

            for i, arg in enumerate(decl["args"]):
                arg["comma"] = "" if i == 0 else ","
                tmp = """{comma}
                        *({type}*)(epc__apistub_buffer + {offset})"""
                f_dispatcher.write(tmp.format(**arg))

            tmp = """
                );
                break;
"""
            f_dispatcher.write(tmp.format(**decl))

        tmp = """\
        default:
                /* TODO: Better error handling */
                LOG_ERR("api-dispatcher", "API function 0x%lx is unknown!!", id);
                break;
        }}
}}
"""
        f_dispatcher.write(tmp.format(**fmt_header))
    # END: Generate Dispatcher }}}


if __name__ == "__main__":
    main()