Source code for datawork.api.tool

"""Module implementing Tool and Invocation."""

import hashlib
from abc import abstractstaticmethod

from .cache import Hashable
from .config import Configurable, Option
from .data import Data
from .invocation import Invocation


class Tool(Configurable, Hashable):
    """
    A class for composable tools.

    This is the base class for configurable functions that transform
    :class:`~.data.Data` objects.
    """

    # Each of these should be lists of placeholder objects that subclass 'Data'
    INPUTS = []
    OUTPUTS = []

    def __init__(self):
        """Construct an object with a new configuration."""
        Configurable.__init__(self)
        Hashable.__init__(self)

    def __repr__(self):
        """Return the name of the tool and its config."""
        return self.__class__.__name__ + ": config=" + repr(self.config)

    def version(self):
        """
        Return a version string for this tool.

        Subclasses implement their own version methods.
        """
        raise NotImplementedError

    def get_hash(self):
        """Return a hash string for this tool and configuration."""
        return ""
        # TODO: add tool version identifier
        config_hash = self.config.get_hash()

        return hashlib.md5(("tool" + config_hash).encode("ascii")).hexdigest()

    @abstractstaticmethod
    def run(cfg):
        """
        Compute outputs.

        This method is the meat of the :class:`~.tool.Tool` class.
        """
        pass

    def __call__(self, *args):
        """
        Let a tool act on some :class:`~.data.Data` objects.

        Calling a tool instance _invokes_ the tool, which results in an
        :class:`~.invocation.Invocation` instance if the arguments are
        of class :class:`~.data.Data`.
        """
        newargs = []
        for ii, a in zip(self.INPUTS, args):
            if a is None:  # create a new placeholder object
                a = ii.__class__()
            assert isinstance(
                a, Data
            ), "__call__() expects all arguments to be instances of Data or None"
            newargs.append(a)
        return Invocation(self, tuple(newargs)).o


def tool(r):
    """
    Quickly create a :class:`~.tool.Tool` class.

    Given a function definition for a tool, create a class with the
    provided function as the class's :meth:`~.tool.Tool.run` method.
    """
    import inspect

    ins = []
    outs = []
    opts = []

    sig = inspect.signature(r)
    for arg, p in sig.parameters.items():
        ann = p.annotation
        if isinstance(ann, Data):
            ann.name = arg  # monkey patch the name onto the annotation
            ins.append(ann)
        elif isinstance(ann, Option):
            ann.name = arg
            if p.default is p.empty:
                ann.default = p.default
            opts.append(ann)
        else:
            print(arg)
            raise Exception("All arguments must be annotated")

    # handle return annotation
    ann = sig.return_annotation
    if ann is inspect.Parameter.empty:
        raise Exception("Return values must be annotated")
    if not isinstance(ann, (list, tuple)):
        ann = [ann]
    for oi in ann:
        outs.append(oi)

    return type(r.__name__, (Tool,), dict(run=r, __doc__=r.__doc__,
                                          INPUTS=ins, OUTPUTS=outs, OPTIONS=opts))