path: root/rtemsqual/interface.py

# SPDX-License-Identifier: BSD-2-Clause
""" This module provides functions for the generation of interfaces. """

# Copyright (C) 2020 embedded brains GmbH (http://www.embedded-brains.de)
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
# 1. Redistributions of source code must retain the above copyright
#    notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
#    notice, this list of conditions and the following disclaimer in the
#    documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

from contextlib import contextmanager
import os
from typing import Any, Callable, Dict, Iterator, List, Optional, Union

from rtemsqual.content import CContent, CInclude, enabled_by_to_exp, \
    ExpressionMapper
from rtemsqual.items import Item, ItemCache, ItemMapper

ItemMap = Dict[str, Item]
Lines = Union[str, List[str]]
GetLines = Callable[["Node", Item, Any], Lines]


def _get_ingroups(item: Item) -> ItemMap:
    ingroups = {}  # type: ItemMap
    for link in item.links_to_parents():
        if link.role == "interface-ingroup":
            ingroups[link.item.uid] = link.item
    return ingroups


def _get_group_identifiers(groups: ItemMap) -> List[str]:
    return [item["identifier"] for item in groups.values()]


def _forward_declaration(item: Item) -> str:
    target = next(item.parents("interface-target"))
    return f"{target['interface-type']} {target['name']}"


class _InterfaceMapper(ItemMapper):
    def __init__(self, node: "Node"):
        super().__init__(node.item)
        self._node = node

    def __getitem__(self, identifier):
        item, value = self.map(identifier)
        if item["type"] == "interface":
            node = self._node
            header_file = node.header_file
            if item["interface-type"] == "enumerator":
                for link in item.links_to_children():
                    if link.role == "interface-enumerator":
                        header_file.add_includes(link.item)
            else:
                header_file.add_includes(item)
            header_file.add_potential_edge(node, item)
        return value

    def get_value(self, item: Item, _path: str, _value: Any, key: str,
                  _index: Optional[int]) -> Any:
        # pylint: disable=no-self-use
        if key == "name" and item["type"] == "interface" and item[
                "interface-type"] == "forward-declaration":
            return _forward_declaration(item)
        raise KeyError

    def enabled_by_to_defined(self, enabled_by: str) -> str:
        """
        Maps an item-level enabled-by attribute value to the corresponding
        defined expression.
        """
        return self._node.header_file.enabled_by_defined[enabled_by]


class _InterfaceExpressionMapper(ExpressionMapper):
    def __init__(self, mapper: _InterfaceMapper):
        super().__init__()
        self._mapper = mapper

    def map_symbol(self, symbol: str) -> str:
        return self._mapper.substitute(symbol)


class _ItemLevelExpressionMapper(ExpressionMapper):
    def __init__(self, mapper: _InterfaceMapper):
        super().__init__()
        self._mapper = mapper

    def map_symbol(self, symbol: str) -> str:
        return self._mapper.substitute(
            self._mapper.enabled_by_to_defined(symbol))


class _HeaderExpressionMapper(ExpressionMapper):
    def __init__(self, item: Item, enabled_by_defined: Dict[str, str]):
        super().__init__()
        self._mapper = ItemMapper(item)
        self._enabled_by_defined = enabled_by_defined

    def map_symbol(self, symbol: str) -> str:
        return self._mapper.substitute(self._enabled_by_defined[symbol])


def _add_definition(node: "Node", item: Item, prefix: str,
                    value: Dict[str, Any], get_lines: GetLines) -> CContent:
    content = CContent()
    default = value["default"]
    variants = value["variants"]
    if variants:
        ifelse = "#if "
        with node.mapper.prefix(os.path.join(prefix, "variants")):
            for variant in variants:
                enabled_by = enabled_by_to_exp(
                    variant["enabled-by"],
                    _InterfaceExpressionMapper(node.mapper))
                content.append(f"{ifelse}{enabled_by}")
                with content.indent():
                    content.append(get_lines(node, item,
                                             variant["definition"]))
                ifelse = "#elif "
        if default is not None:
            content.append("#else")
            with node.mapper.prefix(os.path.join(prefix, "default")):
                with content.indent():
                    content.append(get_lines(node, item, default))
        content.append("#endif")
    else:
        with node.mapper.prefix(os.path.join(prefix, "default")):
            content.append(get_lines(node, item, default))
    return content


_PARAM = {
    None: "@param ",
    "in": "@param[in] ",
    "out": "@param[out] ",
    "inout": "@param[in,out] ",
}


class Node:
    """ Nodes of a header file. """
    def __init__(self, header_file: "_HeaderFile", item: Item,
                 ingroups: ItemMap):
        self.header_file = header_file
        self.item = item
        self.ingroups = ingroups
        self.in_edges = {}  # type: ItemMap
        self.out_edges = {}  # type: ItemMap
        self.content = CContent()
        self.mapper = _InterfaceMapper(self)

    def __lt__(self, other: "Node") -> bool:
        return self.item.uid < other.item.uid

    @contextmanager
    def _enum_struct_or_union(self) -> Iterator[None]:
        self.content.add(self._get_description(self.item, self.ingroups))
        name = self.item["name"]
        typename = self.item["interface-type"]
        kind = self.item["definition-kind"]
        if kind == f"{typename}-only":
            self.content.append(f"{typename} {name} {{")
        elif kind == "typedef-only":
            self.content.append(f"typedef {typename} {{")
        else:
            self.content.append(f"typedef {typename} {name} {{")
        self.content.push_indent()
        yield
        self.content.pop_indent()
        if kind == f"{typename}-only":
            self.content.append("};")
        else:
            self.content.append(f"}} {name};")

    def _generate(self) -> None:
        _NODE_GENERATORS[self.item["interface-type"]](self)

    def generate(self) -> None:
        """ Generates a node to generate the node content. """
        enabled_by = self.item["enabled-by"]
        if not isinstance(enabled_by, bool) or not enabled_by:
            mapper = _ItemLevelExpressionMapper(self.mapper)
            self.content.add(f"#if {enabled_by_to_exp(enabled_by, mapper)}")
            with self.content.indent():
                self._generate()
            self.content.add("#endif")
        else:
            self._generate()

    def generate_compound(self) -> None:
        """ Generates a compound (struct or union). """
        with self._enum_struct_or_union():
            index = 0
            for definition in self.item["definition"]:
                self.content.add(
                    _add_definition(self, self.item, f"definition[{index}]",
                                    definition, Node._get_compound_definition))
                index += 1

    def generate_enum(self) -> None:
        """ Generates an enum. """
        with self._enum_struct_or_union():
            enumerators = []  # type: List[CContent]
            for link in self.item.links_to_parents():
                if link.role != "interface-enumerator":
                    continue
                enumerator = self._get_description(link.item, {})
                enumerator.append(
                    _add_definition(self, link.item, "definition",
                                    link.item["definition"],
                                    Node._get_enumerator_definition))
                enumerators.append(enumerator)
            for enumerator in enumerators[0:-1]:
                enumerator.lines[-1] += ","
                enumerator.append("")
                self.content.append(enumerator)
            try:
                self.content.append(enumerators[-1])
            except IndexError:
                pass

    def generate_define(self) -> None:
        """ Generates a define. """
        self._add_generic_definition(Node._get_define_definition)

    def generate_forward_declaration(self) -> None:
        """ Generates a forward declaration. """
        self.content.append([
            "", "/* Forward declaration */",
            _forward_declaration(self.item) + ";"
        ])

    def generate_function(self) -> None:
        """ Generates a function. """
        self._add_generic_definition(Node._get_function_definition)

    def generate_group(self) -> None:
        """ Generates a group. """
        self.header_file.add_ingroup(self.item)
        for ingroup in self.ingroups.values():
            self.header_file.add_potential_edge(self, ingroup)
        self.content.add_group(self.item["identifier"], self.item["name"],
                               _get_group_identifiers(self.ingroups),
                               self.item["brief"], self.item["description"])

    def generate_macro(self) -> None:
        """ Generates a macro. """
        self._add_generic_definition(Node._get_macro_definition)

    def generate_typedef(self) -> None:
        """ Generates a typedef. """
        self._add_generic_definition(Node._get_typedef_definition)

    def generate_variable(self) -> None:
        """ Generates a variable. """
        self._add_generic_definition(Node._get_variable_definition)

    def substitute(self, text: str) -> str:
        """
        Performs a variable substitution using the item mapper of the node.
        """
        if text:
            return self.mapper.substitute(text.strip("\n"))
        return text

    def _get_compound_definition(self, item: Item, definition: Any) -> Lines:
        content = CContent()
        with content.doxygen_block():
            content.add_brief_description(self.substitute(definition["brief"]))
            content.wrap(self.substitute(definition["description"]))
        kind = definition["kind"]
        if kind == "member":
            member = self.substitute(definition["definition"]) + ";"
            content.append(member.split("\n"))
        else:
            content.append(f"{kind} {{")
            content.gap = False
            with content.indent():
                index = 0
                for compound_member in definition["definition"]:
                    content.add(
                        _add_definition(self, item, f"definition[{index}]",
                                        compound_member,
                                        Node._get_compound_definition))
                    index += 1
            name = definition["name"]
            content.append(f"}} {name};")
        return content.lines

    def _get_enumerator_definition(self, item: Item, definition: Any) -> Lines:
        name = item["name"]
        if definition:
            return f"{name} = {self.substitute(definition)}"
        return f"{name}"

    def _get_define_definition(self, item: Item, definition: Any) -> Lines:
        name = item["name"]
        value = self.substitute(definition)
        if value:
            return f"#define {name} {value}".split("\n")
        return f"#define {name}"

    def _get_function_definition(self, item: Item, definition: Any) -> Lines:
        body = definition["body"]
        ret = self.substitute(definition["return"])
        if body:
            ret = "static inline " + ret
        name = item["name"]
        space = "" if ret.endswith("*") else " "
        if definition["params"]:
            params = [self.substitute(param) for param in definition["params"]]
            param_line = ", ".join(params)
            line = f"{ret}{space}{name}({param_line})"
            if len(line) > 79:
                line = f"{ret}{space}{name}"
                param_block = ",\n  ".join(params)
                line = f"{ret}{space}{name}(\n  {param_block}\n)"
        else:
            line = f"{ret}{space}{name}(void)"
            if len(line) > 79:
                line = f"{ret}\n{name}(void)"
        if body:
            body_lines = self.substitute("\n  ".join(
                body.strip("\n").split("\n")))
            line = f"""{line}
{{
  {body_lines}
}}"""
        else:
            line += ";"
        return line

    def _get_macro_definition(self, item: Item, definition: Any) -> Lines:
        name = item["name"]
        params = [param["name"] for param in item["params"]]
        param_line = ", ".join(params)
        line = f"#define {name}({param_line})"
        if len(line) > 79:
            param_block = ", \\\n  ".join(params)
            line = f"#define {name}( \\\n  {param_block} \\\n)"
        if not definition:
            return line
        body_lines = definition.strip("\n").split("\n")
        if len(body_lines) == 1 and len(line + body_lines[0]) <= 79:
            body = " "
        else:
            body = " \\\n  "
        body += self.substitute(" \\\n  ".join(body_lines))
        return line + body

    def _get_typedef_definition(self, _item: Item, definition: Any) -> Lines:
        return f"typedef {self.substitute(definition)};"

    def _get_variable_definition(self, _item: Item, definition: Any) -> Lines:
        return f"extern {self.substitute(definition)};"

    def _get_description(self, item: Item, ingroups: ItemMap) -> CContent:
        content = CContent()
        with content.doxygen_block():
            content.add_ingroup(_get_group_identifiers(ingroups))
            content.add_brief_description(self.substitute(item["brief"]))
            content.wrap(self.substitute(item["description"]))
            content.wrap(self.substitute(item["notes"]))
            if "params" in item:
                for param in item["params"]:
                    content.wrap(param["name"] + " " +
                                 self.substitute(param["description"]),
                                 initial_indent=_PARAM[param["dir"]])
            if "return" in item:
                ret = item["return"]
                for retval in ret["return-values"]:
                    content.wrap(self.substitute(retval["description"]),
                                 initial_indent=f"@retval {retval['value']} ")
                content.wrap(ret["return"], initial_indent="@return ")
        return content

    def _add_generic_definition(self, get_lines: GetLines) -> None:
        self.content.add(self._get_description(self.item, self.ingroups))
        self.content.append(
            _add_definition(self, self.item, "definition",
                            self.item["definition"], get_lines))


_NODE_GENERATORS = {
    "enum": Node.generate_enum,
    "define": Node.generate_define,
    "forward-declaration": Node.generate_forward_declaration,
    "function": Node.generate_function,
    "group": Node.generate_group,
    "macro": Node.generate_macro,
    "struct": Node.generate_compound,
    "typedef": Node.generate_typedef,
    "union": Node.generate_compound,
    "variable": Node.generate_variable,
}


class _HeaderFile:
    """ A header file. """
    def __init__(self, item: Item, enabled_by_defined: Dict[str, str]):
        self._item = item
        self._content = CContent()
        self._ingroups = {}  # type: ItemMap
        self._includes = []  # type: List[Item]
        self._nodes = {}  # type: Dict[str, Node]
        self.enabled_by_defined = enabled_by_defined

    def add_includes(self, item: Item) -> None:
        """ Adds the includes of the item to the header file includes. """
        for link in item.links_to_parents():
            if link.role == "interface-placement" and link.item[
                    "interface-type"] == "header-file":
                self._includes.append(link.item)

    def add_ingroup(self, item: Item) -> None:
        """ Adds an ingroup to the header file. """
        self._ingroups[item.uid] = item

    def _add_child(self, item: Item) -> None:
        ingroups = _get_ingroups(item)
        if item["interface-type"] != "group":
            self._ingroups.update(ingroups)
        self._nodes[item.uid] = Node(self, item, ingroups)
        self._content.register_license_and_copyrights_of_item(item)

    def add_potential_edge(self, node: Node, item: Item) -> None:
        """
        Adds a potential edge from a node to another node identified by an
        item.
        """
        if item.uid in self._nodes and item.uid != node.item.uid:
            node.out_edges[item.uid] = item
            self._nodes[item.uid].in_edges[node.item.uid] = node.item

    def _resolve_ingroups(self, node: Node) -> None:
        for ingroup in node.ingroups.values():
            self.add_potential_edge(node, ingroup)

    def generate_nodes(self) -> None:
        """ Generates all nodes of this header file. """
        for link in self._item.links_to_children():
            if link.role == "interface-placement":
                self._add_child(link.item)
        for node in self._nodes.values():
            self._resolve_ingroups(node)
            node.generate()

    def _get_nodes_in_dependency_order(self) -> List[Node]:
        """
        Gets the nodes of this header file ordered according to node
        dependencies and UIDs.

        Performs a topological sort using Kahn's algorithm.
        """
        nodes_in_dependency_order = []  # type: List[Node]

        # Get incoming edge degrees for all nodes
        in_degree = {}  # type: Dict[str, int]
        for node in self._nodes.values():
            in_degree[node.item.uid] = len(node.in_edges)

        # Create a queue with all nodes with no incoming edges sorted by UID
        queue = []  # type: List[Node]
        for node in self._nodes.values():
            if in_degree[node.item.uid] == 0:
                queue.append(node)
        queue.sort(reverse=True)

        # Topological sort
        while queue:
            node = queue.pop(0)
            nodes_in_dependency_order.insert(0, node)

            # Sort by UID
            for uid in sorted(node.out_edges):
                in_degree[uid] -= 1
                if in_degree[uid] == 0:
                    queue.append(self._nodes[uid])

        return nodes_in_dependency_order

    def finalize(self) -> None:
        """ Finalizes the header file. """
        self._content.add_spdx_license_identifier()
        with self._content.file_block():
            self._content.add_ingroup(_get_group_identifiers(self._ingroups))
        self._content.add_copyrights_and_licenses()
        with self._content.header_guard(self._item["path"]):
            exp_mapper = _HeaderExpressionMapper(self._item,
                                                 self.enabled_by_defined)
            includes = [
                CInclude(item["path"],
                         enabled_by_to_exp(item["enabled-by"], exp_mapper))
                for item in self._includes if item != self._item
            ]
            includes.extend([
                CInclude(link.item["path"],
                         enabled_by_to_exp(link["enabled-by"], exp_mapper))
                for link in self._item.links_to_parents()
                if link.role == "interface-include"
            ])
            self._content.add_includes(includes)
            with self._content.extern_c():
                for node in self._get_nodes_in_dependency_order():
                    self._content.add(node.content)

    def write(self, domain_path: str) -> None:
        """ Writes the header file. """
        self._content.write(
            os.path.join(domain_path, self._item["prefix"],
                         self._item["path"]))


def _generate_header_file(item: Item, domains: Dict[str, str],
                          enabled_by_defined: Dict[str, str]) -> None:
    domain = next(item.parents("interface-placement"))
    assert domain["interface-type"] == "domain"
    domain_path = domains.get(domain.uid, None)
    if domain_path is None:
        return
    header_file = _HeaderFile(item, enabled_by_defined)
    header_file.generate_nodes()
    header_file.finalize()
    header_file.write(domain_path)


def _visit_header_files(item: Item, domains: Dict[str, str],
                        enabled_by_defined: Dict[str, str]) -> None:
    for child in item.children():
        _visit_header_files(child, domains, enabled_by_defined)
    if item["type"] == "interface" and item["interface-type"] == "header-file":
        _generate_header_file(item, domains, enabled_by_defined)


def _gather_enabled_by_defined(item_level_interfaces: List[str],
                               item_cache: ItemCache) -> Dict[str, str]:
    enabled_by_defined = {}  # type: Dict[str, str]
    for uid in item_level_interfaces:
        for link in item_cache[uid].links_to_children():
            if link.role == "interface-placement":
                define = f"defined(${{{link.item.uid}:/name}})"
                enabled_by_defined[link.item["name"]] = define
    return enabled_by_defined


def generate(config: dict, item_cache: ItemCache) -> None:
    """
    Generates header files according to the configuration.

    :param config: A dictionary with configuration entries.
    :param item_cache: The specification item cache containing the interfaces.
    """
    enabled_by_defined = _gather_enabled_by_defined(
        config["item-level-interfaces"], item_cache)
    for item in item_cache.top_level.values():
        _visit_header_files(item, config["domains"], enabled_by_defined)