pimgen.py

#!/usr/bin/env python3

"""Phosphor Inventory Manager YAML parser and code generator.

The parser workflow is broken down as follows:
  1 - Import YAML files as native python type(s) instance(s).
  2 - Create an instance of the Everything class from the
        native python type instance(s) with the Everything.load
        method.
  3 - The Everything class constructor orchestrates conversion of the
        native python type(s) instances(s) to render helper types.
        Each render helper type constructor imports its attributes
        from the native python type(s) instances(s).
  4 - Present the converted YAML to the command processing method
        requested by the script user.
"""

import argparse
import os
import sys

import mako.lookup
import sdbusplus.property
import yaml
from sdbusplus.namedelement import NamedElement
from sdbusplus.renderer import Renderer

# Global busname for use within classes where necessary
busname = "xyz.openbmc_project.Inventory.Manager"


def cppTypeName(yaml_type):
    """Convert yaml types to cpp types."""
    return sdbusplus.property.Property(type=yaml_type).cppTypeName


class InterfaceComposite(object):
    """Compose interface properties."""

    def __init__(self, dict):
        self.dict = dict

    def interfaces(self):
        return list(self.dict.keys())

    def names(self, interface):
        names = []
        if self.dict[interface]:
            names = [
                NamedElement(name=x["name"]) for x in self.dict[interface]
            ]
        return names


class Interface(list):
    """Provide various interface transformations."""

    def __init__(self, iface):
        super(Interface, self).__init__(iface.split("."))

    def namespace(self):
        """Represent as an sdbusplus namespace."""
        return "::".join(["sdbusplus"] + self[:-1] + ["server", self[-1]])

    def header(self):
        """Represent as an sdbusplus server binding header."""
        return os.sep.join(self + ["server.hpp"])

    def __str__(self):
        return ".".join(self)


class Indent(object):
    """Help templates be depth agnostic."""

    def __init__(self, depth=0):
        self.depth = depth

    def __add__(self, depth):
        return Indent(self.depth + depth)

    def __call__(self, depth):
        """Render an indent at the current depth plus depth."""
        return 4 * " " * (depth + self.depth)


class Template(NamedElement):
    """Associate a template name with its namespace."""

    def __init__(self, **kw):
        self.namespace = kw.pop("namespace", [])
        super(Template, self).__init__(**kw)

    def qualified(self):
        return "::".join(self.namespace + [self.name])


class FixBool(object):
    """Un-capitalize booleans."""

    def __call__(self, arg):
        return "{0}".format(arg.lower())


class Quote(object):
    """Decorate an argument by quoting it."""

    def __call__(self, arg):
        return '"{0}"'.format(arg)


class Cast(object):
    """Decorate an argument by casting it."""

    def __init__(self, cast, target):
        """cast is the cast type (static, const, etc...).
        target is the cast target type."""
        self.cast = cast
        self.target = target

    def __call__(self, arg):
        return "{0}_cast<{1}>({2})".format(self.cast, self.target, arg)


class Literal(object):
    """Decorate an argument with a literal operator."""

    integer_types = [
        "int8",
        "int16",
        "int32",
        "int64",
        "uint8",
        "uint16",
        "uint32",
        "uint64",
    ]

    def __init__(self, type):
        self.type = type

    def __call__(self, arg):
        if "uint" in self.type:
            arg = "{0}ull".format(arg)
        elif "int" in self.type:
            arg = "{0}ll".format(arg)

        if self.type in self.integer_types:
            return Cast("static", "{0}_t".format(self.type))(arg)

        if self.type == "string":
            return "{0}s".format(arg)

        return arg


class Argument(NamedElement, Renderer):
    """Define argument type inteface."""

    def __init__(self, **kw):
        self.type = kw.pop("type", None)
        super(Argument, self).__init__(**kw)

    def argument(self, loader, indent):
        raise NotImplementedError


class TrivialArgument(Argument):
    """Non-array type arguments."""

    def __init__(self, **kw):
        self.value = kw.pop("value")
        self.decorators = kw.pop("decorators", [])
        if kw.get("type", None) == "string":
            self.decorators.insert(0, Quote())
        if kw.get("type", None) == "boolean":
            self.decorators.insert(0, FixBool())

        super(TrivialArgument, self).__init__(**kw)

    def argument(self, loader, indent):
        a = str(self.value)
        for d in self.decorators:
            a = d(a)

        return a


class InitializerList(Argument):
    """Initializer list arguments."""

    def __init__(self, **kw):
        self.values = kw.pop("values")
        super(InitializerList, self).__init__(**kw)

    def argument(self, loader, indent):
        return self.render(
            loader, "argument.mako.cpp", arg=self, indent=indent
        )


class DbusSignature(Argument):
    """DBus signature arguments."""

    def __init__(self, **kw):
        self.sig = {x: y for x, y in kw.items()}
        kw.clear()
        super(DbusSignature, self).__init__(**kw)

    def argument(self, loader, indent):
        return self.render(
            loader, "signature.mako.cpp", signature=self, indent=indent
        )


class MethodCall(Argument):
    """Render syntatically correct c++ method calls."""

    def __init__(self, **kw):
        self.namespace = kw.pop("namespace", [])
        self.templates = kw.pop("templates", [])
        self.args = kw.pop("args", [])
        super(MethodCall, self).__init__(**kw)

    def call(self, loader, indent):
        return self.render(
            loader, "method.mako.cpp", method=self, indent=indent
        )

    def argument(self, loader, indent):
        return self.call(loader, indent)


class Vector(MethodCall):
    """Convenience type for vectors."""

    def __init__(self, **kw):
        kw["name"] = "vector"
        kw["namespace"] = ["std"]
        kw["args"] = [InitializerList(values=kw.pop("args"))]
        super(Vector, self).__init__(**kw)


class Filter(MethodCall):
    """Convenience type for filters"""

    def __init__(self, **kw):
        kw["name"] = "make_filter"
        super(Filter, self).__init__(**kw)


class Action(MethodCall):
    """Convenience type for actions"""

    def __init__(self, **kw):
        kw["name"] = "make_action"
        super(Action, self).__init__(**kw)


class PathCondition(MethodCall):
    """Convenience type for path conditions"""

    def __init__(self, **kw):
        kw["name"] = "make_path_condition"
        super(PathCondition, self).__init__(**kw)


class GetProperty(MethodCall):
    """Convenience type for getting inventory properties"""

    def __init__(self, **kw):
        kw["name"] = "make_get_property"
        super(GetProperty, self).__init__(**kw)


class CreateObjects(MethodCall):
    """Assemble a createObjects functor."""

    def __init__(self, **kw):
        objs = []

        for path, interfaces in kw.pop("objs").items():
            key_o = TrivialArgument(
                value=path, type="string", decorators=[Literal("string")]
            )
            value_i = []

            for (
                interface,
                properties,
            ) in interfaces.items():
                key_i = TrivialArgument(value=interface, type="string")
                value_p = []
                if properties:
                    for prop, value in properties.items():
                        key_p = TrivialArgument(value=prop, type="string")
                        value_v = TrivialArgument(
                            decorators=[Literal(value.get("type", None))],
                            **value
                        )
                        value_p.append(
                            InitializerList(values=[key_p, value_v])
                        )

                value_p = InitializerList(values=value_p)
                value_i.append(InitializerList(values=[key_i, value_p]))

            value_i = InitializerList(values=value_i)
            objs.append(InitializerList(values=[key_o, value_i]))

        kw["args"] = [InitializerList(values=objs)]
        kw["namespace"] = ["functor"]
        super(CreateObjects, self).__init__(**kw)


class DestroyObjects(MethodCall):
    """Assemble a destroyObject functor."""

    def __init__(self, **kw):
        values = [{"value": x, "type": "string"} for x in kw.pop("paths")]
        conditions = [
            Event.functor_map[x["name"]](**x) for x in kw.pop("conditions", [])
        ]
        conditions = [PathCondition(args=[x]) for x in conditions]
        args = [InitializerList(values=[TrivialArgument(**x) for x in values])]
        args.append(InitializerList(values=conditions))
        kw["args"] = args
        kw["namespace"] = ["functor"]
        super(DestroyObjects, self).__init__(**kw)


class SetProperty(MethodCall):
    """Assemble a setProperty functor."""

    def __init__(self, **kw):
        args = []

        value = kw.pop("value")
        prop = kw.pop("property")
        iface = kw.pop("interface")
        iface = Interface(iface)
        namespace = iface.namespace().split("::")[:-1]
        name = iface[-1]
        t = Template(namespace=namespace, name=iface[-1])

        member = "&%s" % "::".join(
            namespace + [name, NamedElement(name=prop).camelCase]
        )
        member_type = cppTypeName(value["type"])
        member_cast = "{0} ({1}::*)({0})".format(member_type, t.qualified())

        paths = [{"value": x, "type": "string"} for x in kw.pop("paths")]
        args.append(
            InitializerList(values=[TrivialArgument(**x) for x in paths])
        )

        conditions = [
            Event.functor_map[x["name"]](**x) for x in kw.pop("conditions", [])
        ]
        conditions = [PathCondition(args=[x]) for x in conditions]

        args.append(InitializerList(values=conditions))
        args.append(TrivialArgument(value=str(iface), type="string"))
        args.append(
            TrivialArgument(
                value=member, decorators=[Cast("static", member_cast)]
            )
        )
        args.append(TrivialArgument(**value))

        kw["templates"] = [Template(name=name, namespace=namespace)]
        kw["args"] = args
        kw["namespace"] = ["functor"]
        super(SetProperty, self).__init__(**kw)


class PropertyChanged(MethodCall):
    """Assemble a propertyChanged functor."""

    def __init__(self, **kw):
        args = []
        args.append(TrivialArgument(value=kw.pop("interface"), type="string"))
        args.append(TrivialArgument(value=kw.pop("property"), type="string"))
        args.append(
            TrivialArgument(
                decorators=[Literal(kw["value"].get("type", None))],
                **kw.pop("value")
            )
        )
        kw["args"] = args
        kw["namespace"] = ["functor"]
        super(PropertyChanged, self).__init__(**kw)


class PropertyIs(MethodCall):
    """Assemble a propertyIs functor."""

    def __init__(self, **kw):
        args = []
        path = kw.pop("path", None)
        if not path:
            path = TrivialArgument(value="nullptr")
        else:
            path = TrivialArgument(value=path, type="string")

        args.append(path)
        iface = TrivialArgument(value=kw.pop("interface"), type="string")
        args.append(iface)
        prop = TrivialArgument(value=kw.pop("property"), type="string")
        args.append(prop)
        args.append(
            TrivialArgument(
                decorators=[Literal(kw["value"].get("type", None))],
                **kw.pop("value")
            )
        )

        service = kw.pop("service", None)
        if service:
            args.append(TrivialArgument(value=service, type="string"))

        dbusMember = kw.pop("dbusMember", None)
        if dbusMember:
            # Inventory manager's service name is required
            if not service or service != busname:
                args.append(TrivialArgument(value=busname, type="string"))

            gpArgs = []
            gpArgs.append(path)
            gpArgs.append(iface)
            # Prepend '&' and append 'getPropertyByName' function on dbusMember
            gpArgs.append(
                TrivialArgument(value="&" + dbusMember + "::getPropertyByName")
            )
            gpArgs.append(prop)
            fArg = MethodCall(
                name="getProperty",
                namespace=["functor"],
                templates=[Template(name=dbusMember, namespace=[])],
                args=gpArgs,
            )

            # Append getProperty functor
            args.append(
                GetProperty(
                    templates=[
                        Template(
                            name=dbusMember + "::PropertiesVariant",
                            namespace=[],
                        )
                    ],
                    args=[fArg],
                )
            )

        kw["args"] = args
        kw["namespace"] = ["functor"]
        super(PropertyIs, self).__init__(**kw)


class Event(MethodCall):
    """Assemble an inventory manager event."""

    functor_map = {
        "destroyObjects": DestroyObjects,
        "createObjects": CreateObjects,
        "propertyChangedTo": PropertyChanged,
        "propertyIs": PropertyIs,
        "setProperty": SetProperty,
    }

    def __init__(self, **kw):
        self.summary = kw.pop("name")

        filters = [
            self.functor_map[x["name"]](**x) for x in kw.pop("filters", [])
        ]
        filters = [Filter(args=[x]) for x in filters]
        filters = Vector(
            templates=[Template(name="Filter", namespace=[])], args=filters
        )

        event = MethodCall(
            name="make_shared",
            namespace=["std"],
            templates=[
                Template(
                    name=kw.pop("event"),
                    namespace=kw.pop("event_namespace", []),
                )
            ],
            args=kw.pop("event_args", []) + [filters],
        )

        events = Vector(
            templates=[Template(name="EventBasePtr", namespace=[])],
            args=[event],
        )

        action_type = Template(name="Action", namespace=[])
        action_args = [
            self.functor_map[x["name"]](**x) for x in kw.pop("actions", [])
        ]
        action_args = [Action(args=[x]) for x in action_args]
        actions = Vector(templates=[action_type], args=action_args)

        kw["name"] = "make_tuple"
        kw["namespace"] = ["std"]
        kw["args"] = [events, actions]
        super(Event, self).__init__(**kw)


class MatchEvent(Event):
    """Associate one or more dbus signal match signatures with
    a filter."""

    def __init__(self, **kw):
        kw["event"] = "DbusSignal"
        kw["event_namespace"] = []
        kw["event_args"] = [
            DbusSignature(**x) for x in kw.pop("signatures", [])
        ]

        super(MatchEvent, self).__init__(**kw)


class StartupEvent(Event):
    """Assemble a startup event."""

    def __init__(self, **kw):
        kw["event"] = "StartupEvent"
        kw["event_namespace"] = []
        super(StartupEvent, self).__init__(**kw)


class Everything(Renderer):
    """Parse/render entry point."""

    class_map = {
        "match": MatchEvent,
        "startup": StartupEvent,
    }

    @staticmethod
    def load(args):
        # Aggregate all the event YAML in the events.d directory
        # into a single list of events.

        events = []
        events_dir = os.path.join(args.inputdir, "events.d")

        if os.path.exists(events_dir):
            yaml_files = [
                x for x in os.listdir(events_dir) if x.endswith(".yaml")
            ]

            for x in yaml_files:
                with open(os.path.join(events_dir, x), "r") as fd:
                    for e in yaml.safe_load(fd.read()).get("events", {}):
                        events.append(e)

        interfaces, interface_composite = Everything.get_interfaces(
            args.ifacesdir
        )
        (
            extra_interfaces,
            extra_interface_composite,
        ) = Everything.get_interfaces(
            os.path.join(args.inputdir, "extra_interfaces.d")
        )
        interface_composite.update(extra_interface_composite)
        interface_composite = InterfaceComposite(interface_composite)
        # Update busname if configured differenly than the default
        global busname
        busname = args.busname

        return Everything(
            *events,
            interfaces=interfaces + extra_interfaces,
            interface_composite=interface_composite
        )

    @staticmethod
    def get_interfaces(targetdir):
        """Scan the interfaces directory for interfaces that PIM can create."""

        yaml_files = []
        interfaces = []
        interface_composite = {}

        if targetdir and os.path.exists(targetdir):
            for directory, _, files in os.walk(targetdir):
                if not files:
                    continue

                yaml_files += [
                    os.path.relpath(os.path.join(directory, f), targetdir)
                    for f in [
                        f for f in files if f.endswith(".interface.yaml")
                    ]
                ]

        for y in yaml_files:
            # parse only phosphor dbus related interface files
            if not (
                y.startswith("xyz")
                or y.startswith("com/ibm/ipzvpd")
                or y.startswith("com/ibm/Control/Host")
            ):
                continue
            with open(os.path.join(targetdir, y)) as fd:
                i = y.replace(".interface.yaml", "").replace(os.sep, ".")

                # PIM can't create interfaces with methods.
                parsed = yaml.safe_load(fd.read())
                if parsed.get("methods", None):
                    continue
                # Cereal can't understand the type sdbusplus::object_path. This
                # type is a wrapper around std::string. Ignore interfaces
                # having a property of this type for now. The only interface
                # that has a property of this type now is
                # xyz.openbmc_project.Association, which is an unused
                # interface. No inventory objects implement this interface.
                # TODO via openbmc/openbmc#2123 : figure out how to make Cereal
                # understand sdbusplus::object_path.
                properties = parsed.get("properties", None)
                if properties:
                    if any("path" in p["type"] for p in properties):
                        continue
                interface_composite[i] = properties
                interfaces.append(i)

        return interfaces, interface_composite

    def __init__(self, *a, **kw):
        self.interfaces = [Interface(x) for x in kw.pop("interfaces", [])]
        self.interface_composite = kw.pop("interface_composite", {})
        self.events = [self.class_map[x["type"]](**x) for x in a]
        super(Everything, self).__init__(**kw)

    def generate_cpp(self, loader):
        """Render the template with the provided events and interfaces."""
        with open(os.path.join(args.outputdir, "generated.cpp"), "w") as fd:
            fd.write(
                self.render(
                    loader,
                    "generated.mako.cpp",
                    events=self.events,
                    interfaces=self.interfaces,
                    indent=Indent(),
                )
            )

    def generate_serialization(self, loader):
        with open(
            os.path.join(args.outputdir, "gen_serialization.hpp"), "w"
        ) as fd:
            fd.write(
                self.render(
                    loader,
                    "gen_serialization.mako.hpp",
                    interfaces=self.interfaces,
                    interface_composite=self.interface_composite,
                )
            )


if __name__ == "__main__":
    script_dir = os.path.dirname(os.path.realpath(__file__))
    valid_commands = {
        "generate-cpp": "generate_cpp",
        "generate-serialization": "generate_serialization",
    }

    parser = argparse.ArgumentParser(
        description=(
            "Phosphor Inventory Manager (PIM) YAML scanner and code generator."
        )
    )
    parser.add_argument(
        "-o",
        "--output-dir",
        dest="outputdir",
        default=".",
        help="Output directory.",
    )
    parser.add_argument(
        "-i",
        "--interfaces-dir",
        dest="ifacesdir",
        help="Location of interfaces to be supported.",
    )
    parser.add_argument(
        "-d",
        "--dir",
        dest="inputdir",
        default=os.path.join(script_dir, "example"),
        help="Location of files to process.",
    )
    parser.add_argument(
        "-b",
        "--bus-name",
        dest="busname",
        default="xyz.openbmc_project.Inventory.Manager",
        help="Inventory manager busname.",
    )
    parser.add_argument(
        "command",
        metavar="COMMAND",
        type=str,
        choices=list(valid_commands.keys()),
        help="%s." % " | ".join(list(valid_commands.keys())),
    )

    args = parser.parse_args()

    if sys.version_info < (3, 0):
        lookup = mako.lookup.TemplateLookup(
            directories=[script_dir], disable_unicode=True
        )
    else:
        lookup = mako.lookup.TemplateLookup(directories=[script_dir])

    function = getattr(Everything.load(args), valid_commands[args.command])
    function(lookup)