debugging workflow and shell architecture refactoring

Author: tclose

new-docs/source/tutorial/tst.py

@@ -1,26 +1,30 @@
 from pathlib import Path
 from tempfile import mkdtemp
-from fileformats.medimage import Nifti
+import shutil
+from fileformats.medimage import Nifti1
 from pydra.tasks.mrtrix3.v3_0 import MrGrid
+from pydra.utils import user_cache_dir
 
 
 if __name__ == "__main__":
     test_dir = Path(mkdtemp())
 
+    shutil.rmtree(user_cache_dir / "run-cache", ignore_errors=True)
+
     nifti_dir = test_dir / "nifti"
     nifti_dir.mkdir()
 
     for i in range(10):
-        Nifti.sample(
+        Nifti1.sample(
             nifti_dir, seed=i
         )  # Create a dummy NIfTI file in the dest. directory
 
     # Instantiate the task definition, "splitting" over all NIfTI files in the test directory
     # by splitting the "input" input field over all files in the directory
-    mrgrid = MrGrid(voxel=(0.5, 0.5, 0.5)).split(input=nifti_dir.iterdir())
+    mrgrid = MrGrid(voxel=(0.5, 0.5, 0.5)).split(in_file=nifti_dir.iterdir())
 
     # Run the task to resample all NIfTI files
-    outputs = mrgrid()
+    outputs = mrgrid(worker="serial")
 
     # Print the locations of the output files
-    print("\n".join(str(p) for p in outputs.output))
+    print("\n".join(str(p) for p in outputs.outputs))

new-docs/tst.py

@@ -11,4 +11,4 @@
 result = load_json(plugin="serial")
 
 # Print the output interface of the of the task (LoadJson.Outputs)
-print(result.output)
+print(result.outputs)

pydra/design/base.py

@@ -388,7 +388,8 @@ def make_task_def(
     klass : type
         The class created using the attrs package
     """
-    from pydra.engine.specs import TaskDef
+    from pydra.engine.specs import TaskDef, WorkflowDef
+    from pydra.engine.core import Task, WorkflowTask
 
     spec_type._check_arg_refs(inputs, outputs)
 
@@ -399,6 +400,7 @@ def make_task_def(
             f"{reserved_names} are reserved and cannot be used for {spec_type} field names"
         )
     outputs_klass = make_outputs_spec(out_type, outputs, outputs_bases, name)
+    task_type = WorkflowTask if issubclass(spec_type, WorkflowDef) else Task
     if klass is None or not issubclass(klass, spec_type):
         if name is None:
             raise ValueError("name must be provided if klass is not")
@@ -417,13 +419,19 @@ def make_task_def(
             name=name,
             bases=bases,
             kwds={},
-            exec_body=lambda ns: ns.update({"Outputs": outputs_klass}),
+            exec_body=lambda ns: ns.update(
+                {
+                    "Outputs": outputs_klass,
+                    "Task": task_type,
+                }
+            ),
         )
     else:
         # Ensure that the class has it's own annotations dict so we can modify it without
         # messing up other classes
         klass.__annotations__ = copy(klass.__annotations__)
         klass.Outputs = outputs_klass
+        klass.Task = task_type
     # Now that we have saved the attributes in lists to be
     for arg in inputs.values():
         # If an outarg input then the field type should be Path not a FileSet
@@ -769,7 +777,11 @@ def extract_function_inputs_and_outputs(
     type_hints = ty.get_type_hints(function)
     input_types = {}
     input_defaults = {}
+    has_varargs = False
     for p in sig.parameters.values():
+        if p.kind is p.VAR_POSITIONAL or p.kind is p.VAR_KEYWORD:
+            has_varargs = True
+            continue
         input_types[p.name] = type_hints.get(p.name, ty.Any)
         if p.default is not inspect.Parameter.empty:
             input_defaults[p.name] = p.default
@@ -779,11 +791,12 @@ def extract_function_inputs_and_outputs(
                 f"Input names ({inputs}) should not be provided when "
                 "wrapping/decorating a function as "
             )
-        if unrecognised := set(inputs) - set(input_types):
-            raise ValueError(
-                f"Unrecognised input names ({unrecognised}) not present in the signature "
-                f"of the function {function!r}"
-            )
+        if not has_varargs:
+            if unrecognised := set(inputs) - set(input_types):
+                raise ValueError(
+                    f"Unrecognised input names ({unrecognised}) not present in the signature "
+                    f"of the function {function!r}"
+                )
         for inpt_name, type_ in input_types.items():
             try:
                 inpt = inputs[inpt_name]

pydra/design/workflow.py

@@ -170,7 +170,7 @@ def make(wrapped: ty.Callable | type) -> TaskDef:
         for inpt_name in lazy:
             parsed_inputs[inpt_name].lazy = True
 
-        interface = make_task_def(
+        defn = make_task_def(
             WorkflowDef,
             WorkflowOutputs,
             parsed_inputs,
@@ -181,7 +181,7 @@ def make(wrapped: ty.Callable | type) -> TaskDef:
             outputs_bases=outputs_bases,
         )
 
-        return interface
+        return defn
 
     if wrapped is not None:
         if not isinstance(wrapped, (ty.Callable, type)):
@@ -198,7 +198,7 @@ def this() -> "Workflow":
     Workflow
         The workflow currently being constructed.
     """
-    from pydra.engine.workflow.base import Workflow
+    from pydra.engine.core import Workflow
 
     return Workflow.under_construction