From e97aa1292f4c464456c407d1d4e523220a5f65d2 Mon Sep 17 00:00:00 2001 From: Tom Close Date: Mon, 23 Jun 2025 11:05:05 +1000 Subject: [PATCH] created stub for matlab compose --- pydra/compose/matlab/__init__.py | 5 + pydra/compose/matlab/builder.py | 160 +++++++++++++++++++++++++++++++ pydra/compose/matlab/field.py | 64 +++++++++++++ pydra/compose/matlab/task.py | 52 ++++++++++ 4 files changed, 281 insertions(+) create mode 100644 pydra/compose/matlab/__init__.py create mode 100644 pydra/compose/matlab/builder.py create mode 100644 pydra/compose/matlab/field.py create mode 100644 pydra/compose/matlab/task.py diff --git a/pydra/compose/matlab/__init__.py b/pydra/compose/matlab/__init__.py new file mode 100644 index 000000000..c5bab893a --- /dev/null +++ b/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"] diff --git a/pydra/compose/matlab/builder.py b/pydra/compose/matlab/builder.py new file mode 100644 index 000000000..9d6e283c2 --- /dev/null +++ b/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 diff --git a/pydra/compose/matlab/field.py b/pydra/compose/matlab/field.py new file mode 100644 index 000000000..98773e14f --- /dev/null +++ b/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 diff --git a/pydra/compose/matlab/task.py b/pydra/compose/matlab/task.py new file mode 100644 index 000000000..c9c887e83 --- /dev/null +++ b/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