diff --git a/doc/conf.py b/doc/conf.py index df003d0..a7143c6 100644 --- a/doc/conf.py +++ b/doc/conf.py @@ -21,6 +21,7 @@ "sphinx.ext.viewcode", "sphinxcontrib.mermaid", "sphinx_autodoc_typehints", + "sphinx_design", "nbsphinx", "nbsphinx_link", ] diff --git a/doc/howtoguides.rst b/doc/howtoguides.rst index e482333..ab0f329 100644 --- a/doc/howtoguides.rst +++ b/doc/howtoguides.rst @@ -6,5 +6,6 @@ How-to Guides howtoguides/execute_io howtoguides/events howtoguides/task_discovery + howtoguides/function_task howtoguides/notebook_task howtoguides/change_schema diff --git a/doc/howtoguides/function_task.rst b/doc/howtoguides/function_task.rst new file mode 100644 index 0000000..3604545 --- /dev/null +++ b/doc/howtoguides/function_task.rst @@ -0,0 +1,146 @@ +Python function as workflow task +================================ + +A Python function can be used as a task node in a workflow by using ``"method"`` as its ``task_type``. +By default, such task has a single output named ``return_value``. +It is possible to `Define multiple outputs`_. + +Use a function as a task +------------------------ + +Example with a ``range_info`` function returning 3 values in a dictionary: + +.. code:: python + + def range_info(a: float, b: float): + return { + "extent": abs(b - a), + "minimum": min(a, b), + "maximum": max(a, b), + } + +The corresponding workflow node must be declared with ``"method"`` as ``task_type``: + +.. code:: python + + range_info_node = { + "id": "task_range_info", + "task_type": "method", + "task_identifier": "__main__.range_info", + } + +Code to execute the ``range_info`` function as a task in a workflow with ``a=15`` and ``b=10`` as inputs: + +.. code:: python + + from ewokscore import execute_graph + + # Define a workflow which calls the range_info function as a task + workflow = { + "graph": {"id": "range_info_workflow"}, + "nodes": [ + { + "id": "task_range_info", + "task_type": "method", + "task_identifier": "__main__.range_info", + }, + ], + "links": [], + } + + # Define task inputs + inputs = [ + {"id": "task_range_info", "name": "a", "value": 15}, + {"id": "task_range_info", "name": "b", "value": 10}, + ] + + # Execute the workflow + result = execute_graph(workflow, inputs=inputs) + print(result) + +The task output contains a single ``return_value`` field which is set to the function return value. + +In the ``range_info`` example, the result contains a single output field ``return_value``: + +.. code:: python + + {'return_value': {'extent': 5, 'minimum': 10, 'maximum': 15}} + + +Define multiple outputs +----------------------- + +To declare a function that can be used as a :class:`Task` with multiple output fields: + +* Declare the function with one of the following kind of return type: + + .. tab-set:: + + .. tab-item:: namedtuple + + .. code:: python + + from collections import namedtuple + + Result = namedtuple("Result", ["extent", "minimum", "maximum"]) + + .. tab-item:: dataclass + + .. code:: python + + from dataclasses import dataclass + + @dataclass + class Result: + extent: float + minimum: float + maximum: float + + .. tab-item:: typing.NamedTuple + + .. code:: python + + from typing import NamedTuple + + class Result(NamedTuple): + extent: float + minimum: float + maximum: float + + .. tab-item:: typing.TypedDict + + .. code:: python + + from typing import TypedDict + + class Result(TypedDict): + extent: float + minimum: float + maximum: float + + .. code:: python + + def range_info(a: float, b: float) -> Result: + +* Add the :func:`task_outputs` decorator to the function declaration: + + .. code:: python + + from ewokscore.methodtask import task_outputs + + @task_outputs + def range_info(a: float, b: float) -> Result: + return Result( + extent=abs(b - a), + minimum=min(a, b), + maximum=max(a, b), + ) + + +When a ``@task_outputs`` decorated function is used as a :class:`Task`, the task output names are defined by the function return type. + +In the example, the return value of ``range_info`` function is available through 3 task outputs: + +.. code:: python + + {'maximum': 15, 'extent': 5, 'minimum': 10} diff --git a/src/ewokscore/inittask.py b/src/ewokscore/inittask.py index c539301..2cf0781 100644 --- a/src/ewokscore/inittask.py +++ b/src/ewokscore/inittask.py @@ -6,7 +6,7 @@ from ewoksutils.import_utils import import_qualname from .task import Task -from .methodtask import MethodExecutorTask +from .methodtask import get_method_task from .scripttask import ScriptExecutorTask from .ppftasks import PpfMethodExecutorTask from .ppftasks import PpfPortTask @@ -155,8 +155,8 @@ def instantiate_task( return Task.instantiate(task_info["task_identifier"], **task_kwargs) if task_type == "method": - task_inputs[MethodExecutorTask.METHOD_ARGUMENT] = task_info["task_identifier"] - return MethodExecutorTask(**task_kwargs) + task_class = get_method_task(task_info["task_identifier"]) + return task_class(**task_kwargs) if task_type == "ppfmethod": task_inputs[PpfMethodExecutorTask.METHOD_ARGUMENT] = task_info[ @@ -254,7 +254,7 @@ def get_task_class(node_id: NodeIdType, node_attrs: dict): if task_type == "class": return Task.get_subclass(task_info["task_identifier"]) if task_type == "method": - return MethodExecutorTask + return get_method_task(task_info["task_identifier"]) if task_type == "ppfmethod": return PpfMethodExecutorTask if task_type == "ppfport": diff --git a/src/ewokscore/methodtask.py b/src/ewokscore/methodtask.py index 8398ac9..79d7b89 100644 --- a/src/ewokscore/methodtask.py +++ b/src/ewokscore/methodtask.py @@ -1,20 +1,109 @@ -from .task import Task +from collections.abc import Mapping +import functools +import inspect +from types import FunctionType +from typing import get_type_hints, Set + from ewoksutils.import_utils import import_method -METHOD_ARGUMENT = "_method" +from .task import Task +from .utils import is_namedtuple, method_arguments + + +def task_outputs(function: FunctionType) -> FunctionType: + """Function decorator so ewoks extracts task outputs from return type elements/attributes""" + # Report error early + if not _method_output_names(function): + raise ValueError("Function return type does not define any output name") + + function._ewoks_unpack_outputs = True # noqa + return function + + +def _method_output_names(method) -> Set[str]: + sig = inspect.signature(method) + return_type = sig.return_annotation + if return_type is inspect.Signature.empty or not inspect.isclass(return_type): + return set() + + return_annotations = get_type_hints(return_type) + if return_annotations: + return set(return_annotations.keys()) + + if is_namedtuple(return_type): + return set(return_type._fields) + + return set() + +class MethodExecutorTask(Task): + _METHOD_REQUIRED_INPUTS = tuple() + _METHOD_OPTIONAL_INPUTS = tuple() + _METHOD_UNPACK_OUTPUTS = False -class MethodExecutorTask( - Task, input_names=[METHOD_ARGUMENT], output_names=["return_value"] -): - METHOD_ARGUMENT = METHOD_ARGUMENT + def __init_subclass__(cls, task_identifier: str): + cls._TASK_IDENTIFIER = task_identifier + + method = import_method(task_identifier) + cls._METHOD_REQUIRED_INPUTS, cls._METHOD_OPTIONAL_INPUTS = method_arguments( + method + ) + + cls._METHOD_UNPACK_OUTPUTS = getattr(method, "_ewoks_unpack_outputs", False) + if cls._METHOD_UNPACK_OUTPUTS: + output_names = _method_output_names(method) + if not output_names: + raise RuntimeError( + f"{task_identifier}'s return type do not define any output name" + ) + else: + output_names = ["return_value"] + + super().__init_subclass__( + # required inputs are optional to support passing them as positional inputs + optional_input_names=cls._METHOD_REQUIRED_INPUTS + + cls._METHOD_OPTIONAL_INPUTS, + output_names=output_names, + ) + cls.__doc__ = inspect.getdoc(method) + + # Advertise effective method required/optional input names + # rather than the less constrained task ones + + @classmethod + def required_input_names(cls): + return cls._METHOD_REQUIRED_INPUTS + + @classmethod + def optional_input_names(cls): + return cls._METHOD_OPTIONAL_INPUTS def run(self): kwargs = self.get_named_input_values() args = self.get_positional_input_values() - fullname = kwargs.pop(self.METHOD_ARGUMENT) - method = import_method(fullname) + + method = import_method(self._TASK_IDENTIFIER) result = method(*args, **kwargs) - self.outputs.return_value = result + if not self._METHOD_UNPACK_OUTPUTS: + self.outputs.return_value = result + return + + if isinstance(result, Mapping): + for name in self.output_names(): + if name in result: + self.outputs[name] = result[name] + return + + for name in self.output_names(): + if hasattr(result, name): + self.outputs[name] = getattr(result, name) + + +@functools.lru_cache() +def get_method_task(task_identifier: str) -> MethodExecutorTask: + class MethodTask(MethodExecutorTask, task_identifier=task_identifier): + pass + + return MethodTask diff --git a/src/ewokscore/persistence/proxy.py b/src/ewokscore/persistence/proxy.py index cf4f3ad..4df2e59 100644 --- a/src/ewokscore/persistence/proxy.py +++ b/src/ewokscore/persistence/proxy.py @@ -125,6 +125,8 @@ def instantiate( raise ValueError(f"Data proxy scheme '{scheme}' is not supported") def serialize(self) -> Dict[str, str]: + if self.uri is None: + return None return self.uri.serialize() @classmethod diff --git a/src/ewokscore/task_discovery.py b/src/ewokscore/task_discovery.py index 6885ab8..9f7c827 100644 --- a/src/ewokscore/task_discovery.py +++ b/src/ewokscore/task_discovery.py @@ -4,7 +4,7 @@ import logging from fnmatch import fnmatch from types import FunctionType, ModuleType -from typing import Generator, Optional, List, Dict, Tuple, Union +from typing import Generator, Optional, List, Dict, Union if sys.version_info < (3, 9): from importlib_metadata import entry_points as _entry_points @@ -28,7 +28,9 @@ def iter_entry_points(group: str): from ewoksutils.import_utils import qualname from ewoksutils.import_utils import import_module +from .methodtask import get_method_task from .task import Task +from .utils import method_arguments TaskDict = Dict[str, Union[str, List[str]]] @@ -129,9 +131,16 @@ def _iter_method_tasks( if method_name.startswith("_"): continue + task_identifier = qualname(method_qn) + task_class = get_method_task(task_identifier) yield { "task_type": "method", - **_common_method_task_fields(method_name, method_qn, mod), + "task_identifier": task_identifier, + "required_input_names": list(task_class.required_input_names()), + "optional_input_names": list(task_class.optional_input_names()), + "output_names": list(task_class.output_names()), + "category": task_identifier.split(".")[0], + "description": task_class.__doc__, } @@ -157,7 +166,7 @@ def _iter_ppfmethod_tasks( yield { "task_type": "ppfmethod", - **_common_method_task_fields(method_name, method_qn, mod), + **_ppfmethod_task_fields(method_name, method_qn, mod), } @@ -248,30 +257,13 @@ def _onerror(module_name, exception: Optional[Exception] = None): logger.error(f"Module '{module_name}' cannot be imported: {exception}") -def _method_arguments(method) -> Tuple[List[str], List[str]]: - sig = inspect.signature(method) - required_input_names = list() - optional_input_names = list() - for name, param in sig.parameters.items(): - required = param.default is inspect._empty - if param.kind == param.VAR_POSITIONAL: - continue - if param.kind == param.VAR_KEYWORD: - continue - if required: - required_input_names.append(name) - else: - optional_input_names.append(name) - return required_input_names, optional_input_names - - -def _common_method_task_fields( +def _ppfmethod_task_fields( method_name: str, method_qn: FunctionType, mod: ModuleType ) -> Dict[str, Union[str, List[str]]]: task_identifier = qualname(method_qn) method = getattr(mod, method_name) - required_input_names, optional_input_names = _method_arguments(method) + required_input_names, optional_input_names = method_arguments(method) return { "task_identifier": qualname(method_qn), diff --git a/src/ewokscore/tests/test_method_task.py b/src/ewokscore/tests/test_method_task.py index d3f08f7..62c0675 100644 --- a/src/ewokscore/tests/test_method_task.py +++ b/src/ewokscore/tests/test_method_task.py @@ -1,5 +1,18 @@ +import sys +from collections import namedtuple +from dataclasses import dataclass +from typing import NamedTuple + +import pytest + from ewoksutils.import_utils import qualname from ewokscore.task import Task +from ewokscore.methodtask import get_method_task, task_outputs + +if sys.version_info >= (3, 8): + from typing import TypedDict +else: + TypedDict = None def mymethod1(a=0, b=0): @@ -7,11 +20,8 @@ def mymethod1(a=0, b=0): def test_method_task1(varinfo): - task = Task.instantiate( - "MethodExecutorTask", - inputs={"_method": qualname(mymethod1), "a": 3, "b": 5}, - varinfo=varinfo, - ) + task_class = get_method_task(qualname(mymethod1)) + task = task_class(inputs={"a": 3, "b": 5}, varinfo=varinfo) task.execute() assert task.done assert task.get_output_values() == {"return_value": 8} @@ -22,11 +32,8 @@ def mymethod2(*args): def test_method_task2(varinfo): - task = Task.instantiate( - "MethodExecutorTask", - inputs={"_method": qualname(mymethod2), 0: 3, 1: 5}, - varinfo=varinfo, - ) + task_class = get_method_task(qualname(mymethod2)) + task = task_class(inputs={0: 3, 1: 5}, varinfo=varinfo) task.execute() assert task.done assert task.get_output_values() == {"return_value": 8} @@ -42,16 +49,78 @@ def mymethod3(a, *args, b=None, c=3, **kw): def test_method_task3(varinfo): - task = Task.instantiate( - "MethodExecutorTask", - inputs={"_method": qualname(mymethod3), 0: 2, 1: 4, "b": 7, "d": 10}, - varinfo=varinfo, - ) + task_class = get_method_task(qualname(mymethod3)) + task = task_class(inputs={0: 2, 1: 4, "b": 7, "d": 10}, varinfo=varinfo) task.execute() assert task.done assert task.get_output_values() == {"return_value": 26} +@dataclass +class DataClassReturn: + x: int + y: float + + +@task_outputs +def mymethod_dataclass() -> DataClassReturn: + return DataClassReturn(x=1, y=2.5) + + +class TypedNamedTupleReturn(NamedTuple): + x: int + y: float + + +@task_outputs +def mymethod_typed_namedtuple() -> TypedNamedTupleReturn: + return TypedNamedTupleReturn(x=1, y=2.5) + + +NamedTupleReturn = namedtuple("NamedTupleReturn", ["x", "y"]) + + +@task_outputs +def mymethod_namedtuple() -> NamedTupleReturn: + return NamedTupleReturn(x=1, y=2.5) + + +@pytest.mark.parametrize( + "method", [mymethod_dataclass, mymethod_typed_namedtuple, mymethod_namedtuple] +) +def test_method_return(method, varinfo): + task_class = get_method_task(qualname(method)) + task = task_class(inputs=None, varinfo=varinfo) + task.execute() + assert task.done + assert task.get_output_values() == {"x": 1, "y": 2.5} + + +@pytest.mark.skipif( + TypedDict is None, + reason="TypedDict not available for this version of Python", +) +def test_method_return_typeddict(varinfo): + global mymethod_typeddict + + class TypedDictReturn(TypedDict): + x: int + y: float + + @task_outputs + def mymethod_typeddict() -> TypedDictReturn: + return TypedDictReturn(x=1, y=2.5) + + task_class = get_method_task(qualname(mymethod_typeddict)) + task = task_class(inputs={}, varinfo=varinfo) + task.execute() + assert task.done + assert task.get_output_values() == { + "x": 1, + "y": 2.5, + } + + def myppfmethod(a=0, b=0, **kw): return {"a": a + b} diff --git a/src/ewokscore/utils.py b/src/ewokscore/utils.py index de87308..4a0eca4 100644 --- a/src/ewokscore/utils.py +++ b/src/ewokscore/utils.py @@ -1,4 +1,6 @@ +import inspect from collections.abc import Mapping, Sequence +from typing import List, Tuple def dict_merge( @@ -32,3 +34,28 @@ def dict_merge( raise ValueError("Conflict at " + ".".join(_nodes)) else: destination[key] = value + + +def is_namedtuple(type_: type) -> bool: + if not type_.__bases__ == (tuple,) or not hasattr(type_, "_fields"): + return False + if not isinstance(type_._fields, tuple): + return False + return all(isinstance(field, str) for field in type_._fields) + + +def method_arguments(method) -> Tuple[List[str], List[str]]: + sig = inspect.signature(method) + required_input_names = list() + optional_input_names = list() + for name, param in sig.parameters.items(): + required = param.default is inspect._empty + if param.kind == param.VAR_POSITIONAL: + continue + if param.kind == param.VAR_KEYWORD: + continue + if required: + required_input_names.append(name) + else: + optional_input_names.append(name) + return required_input_names, optional_input_names