Draft: "method" task with multiple outputs

doc/conf.py

@@ -21,6 +21,7 @@
     "sphinx.ext.viewcode",
     "sphinxcontrib.mermaid",
     "sphinx_autodoc_typehints",
+    "sphinx_design",
     "nbsphinx",
     "nbsphinx_link",
 ]

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

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}

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":

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

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

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),