[WIP] Implementation of compose module for Matlab tasks

pydra/compose/matlab/__init__.py

@@ -0,0 +1,5 @@
+from .task import Task, Outputs
+from .field import arg, out
+from .builder import define
+
+__all__ = ["arg", "out", "define", "Task", "Outputs"]

pydra/compose/matlab/builder.py

@@ -0,0 +1,160 @@
+import typing as ty
+import inspect
+import re
+from typing import dataclass_transform
+from . import field
+from .task import Task, Outputs
+from pydra.compose.base import (
+    ensure_field_objects,
+    build_task_class,
+    check_explicit_fields_are_none,
+    extract_fields_from_class,
+)
+
+
+@dataclass_transform(
+    kw_only_default=True,
+    field_specifiers=(field.arg,),
+)
+def define(
+    wrapped: type | ty.Callable | None = None,
+    /,
+    inputs: list[str | field.arg] | dict[str, field.arg | type] | None = None,
+    outputs: list[str | field.out] | dict[str, field.out | type] | type | None = None,
+    bases: ty.Sequence[type] = (),
+    outputs_bases: ty.Sequence[type] = (),
+    auto_attribs: bool = True,
+    name: str | None = None,
+    xor: ty.Sequence[str | None] | ty.Sequence[ty.Sequence[str | None]] = (),
+) -> "Task":
+    """
+    Create an interface for a function or a class.
+
+    Parameters
+    ----------
+    wrapped : type | callable | None
+        The function or class to create an interface for.
+    inputs : list[str | Arg] | dict[str, Arg | type] | None
+        The inputs to the function or class.
+    outputs : list[str | base.Out] | dict[str, base.Out | type] | type | None
+        The outputs of the function or class.
+    auto_attribs : bool
+        Whether to use auto_attribs mode when creating the class.
+    name: str | None
+        The name of the returned class
+    xor: Sequence[str | None] | Sequence[Sequence[str | None]], optional
+        Names of args that are exclusive mutually exclusive, which must include
+        the name of the current field. If this list includes None, then none of the
+        fields need to be set.
+
+    Returns
+    -------
+    Task
+        The task class for the Python function
+    """
+
+    def make(wrapped: ty.Callable | type) -> Task:
+        if inspect.isclass(wrapped):
+            klass = wrapped
+            function = klass.function
+            class_name = klass.__name__
+            check_explicit_fields_are_none(klass, inputs, outputs)
+            parsed_inputs, parsed_outputs = extract_fields_from_class(
+                Task,
+                Outputs,
+                klass,
+                field.arg,
+                field.out,
+                auto_attribs,
+                skip_fields=["function"],
+            )
+        else:
+            if not isinstance(wrapped, str):
+                raise ValueError(
+                    f"wrapped must be a class or a string containing a MATLAB snipped, not {wrapped!r}"
+                )
+                klass = None
+                input_helps, output_helps = {}, {}
+
+                function_name, inferred_inputs, inferred_outputs = (
+                    parse_matlab_function(
+                        wrapped,
+                        inputs=inputs,
+                        outputs=outputs,
+                    )
+                )
+
+                parsed_inputs, parsed_outputs = ensure_field_objects(
+                    arg_type=field.arg,
+                    out_type=field.out,
+                    inputs=inferred_inputs,
+                    outputs=inferred_outputs,
+                    input_helps=input_helps,
+                    output_helps=output_helps,
+                )
+
+                if name:
+                    class_name = name
+                else:
+                    class_name = function_name
+                    class_name = re.sub(r"[^\w]", "_", class_name)
+                if class_name[0].isdigit():
+                    class_name = f"_{class_name}"
+
+            # Add in fields from base classes
+            parsed_inputs.update({n: getattr(Task, n) for n in Task.BASE_ATTRS})
+            parsed_outputs.update({n: getattr(Outputs, n) for n in Outputs.BASE_ATTRS})
+
+            function = wrapped
+
+        parsed_inputs["function"] = field.arg(
+            name="function",
+            type=str,
+            default=function,
+            help=Task.FUNCTION_HELP,
+        )
+
+        defn = build_task_class(
+            Task,
+            Outputs,
+            parsed_inputs,
+            parsed_outputs,
+            name=class_name,
+            klass=klass,
+            bases=bases,
+            outputs_bases=outputs_bases,
+            xor=xor,
+        )
+
+        return defn
+
+    if wrapped is not None:
+        if not isinstance(wrapped, (str, type)):
+            raise ValueError(f"wrapped must be a class or a string, not {wrapped!r}")
+        return make(wrapped)
+    return make
+
+
+def parse_matlab_function(
+    function: str,
+    inputs: list[str | field.arg] | dict[str, field.arg | type] | None = None,
+    outputs: list[str | field.out] | dict[str, field.out | type] | type | None = None,
+) -> tuple[str, dict[str, field.arg], dict[str, field.out]]:
+    """
+    Parse a MATLAB function string to extract inputs and outputs.
+
+    Parameters
+    ----------
+    function : str
+        The MATLAB function string.
+    inputs : list or dict, optional
+        The inputs to the function.
+    outputs : list or dict, optional
+        The outputs of the function.
+
+    Returns
+    -------
+    tuple
+        A tuple containing the function name, inferred inputs, and inferred outputs.
+    """
+    raise NotImplementedError

pydra/compose/matlab/field.py

@@ -0,0 +1,64 @@
+import attrs
+from .. import base
+
+
+@attrs.define
+class arg(base.Arg):
+    """Argument of a matlab task
+
+    Parameters
+    ----------
+    help: str
+        A short description of the input field.
+    default : Any, optional
+        the default value for the argument
+    allowed_values: list, optional
+        List of allowed values for the field.
+    requires: list, optional
+        Names of the inputs that are required together with the field.
+    copy_mode: File.CopyMode, optional
+        The mode of copying the file, by default it is File.CopyMode.any
+    copy_collation: File.CopyCollation, optional
+        The collation of the file, by default it is File.CopyCollation.any
+    copy_ext_decomp: File.ExtensionDecomposition, optional
+        The extension decomposition of the file, by default it is
+        File.ExtensionDecomposition.single
+    readonly: bool, optional
+        If True the input field can’t be provided by the user but it aggregates other
+        input fields (for example the fields with argstr: -o {fldA} {fldB}), by default
+        it is False
+    type: type, optional
+        The type of the field, by default it is Any
+    name: str, optional
+        The name of the field, used when specifying a list of fields instead of a mapping
+        from name to field, by default it is None
+    """
+
+    pass
+
+
+@attrs.define
+class out(base.Out):
+    """Output of a Python task
+
+    Parameters
+    ----------
+    name: str, optional
+        The name of the field, used when specifying a list of fields instead of a mapping
+        from name to field, by default it is None
+    type: type, optional
+        The type of the field, by default it is Any
+    help: str, optional
+        A short description of the input field.
+    requires: list, optional
+        Names of the inputs that are required together with the field.
+    converter: callable, optional
+        The converter for the field passed through to the attrs.field, by default it is None
+    validator: callable | iterable[callable], optional
+        The validator(s) for the field passed through to the attrs.field, by default it is None
+    position : int
+        The position of the output in the output list, allows for tuple unpacking of
+        outputs
+    """
+
+    pass

pydra/compose/matlab/task.py

@@ -0,0 +1,52 @@
+import typing as ty
+import attrs
+
+# from pydra.utils.general import get_fields, asdict
+from pydra.compose import base
+
+if ty.TYPE_CHECKING:
+    from pydra.engine.job import Job
+
+
+@attrs.define(kw_only=True, auto_attribs=False, eq=False, repr=False)
+class MatlabOutputs(base.Outputs):
+
+    @classmethod
+    def _from_job(cls, job: "Job[MatlabTask]") -> ty.Self:
+        """Collect the outputs of a job from a combination of the provided inputs,
+        the objects in the output directory, and the stdout and stderr of the process.
+
+        Parameters
+        ----------
+        job : Job[Task]
+            The job whose outputs are being collected.
+        outputs_dict : dict[str, ty.Any]
+            The outputs of the job, as a dictionary
+
+        Returns
+        -------
+        outputs : Outputs
+            The outputs of the job in dataclass
+        """
+        raise NotImplementedError
+
+
+MatlabOutputsType = ty.TypeVar("MatlabOutputsType", bound=MatlabOutputs)
+
+
+@attrs.define(kw_only=True, auto_attribs=False, eq=False, repr=False)
+class MatlabTask(base.Task[MatlabOutputsType]):
+
+    _executor_name = "function"
+
+    FUCNTION_HELP = (
+        "a string containing the definition of the MATLAB function to run in the task"
+    )
+
+    def _run(self, job: "Job[MatlabTask]", rerun: bool = True) -> None:
+        raise NotImplementedError
+
+
+# Alias ShellTask to Task so we can refer to it by shell.Task
+Task = MatlabTask
+Outputs = MatlabOutputs