added cononical workflow test

Author: tclose

pydra/design/base.py

@@ -186,10 +186,14 @@ def __call__(
         return task(**kwargs)
 
     def _check_for_unset_values(self):
-        if unset := [k for k, v in attrs.asdict(self).items() if v is attrs.NOTHING]:
+        if unset := [
+            k
+            for k, v in attrs.asdict(self, recurse=False).items()
+            if v is attrs.NOTHING
+        ]:
             raise ValueError(
-                f"The following values in the {self!r} interface need to be set before it "
-                f"can be executed: {unset}"
+                f"The following values {unset} in the {self!r} interface need to be set "
+                "before the workflow can be constructed"
             )
 
 

pydra/design/shell.py

@@ -413,7 +413,7 @@ def add_arg(name, field_type, kwds, is_option=False):
                 kwds["type"] = field
                 field = field_type(name=name, **kwds)
             elif not isinstance(field, field_type):  # If field type is outarg not out
-                field = field_type(**attrs.asdict(field))
+                field = field_type(**attrs.asdict(field, recurse=False))
             field.name = name
             type_ = kwds.pop("type", field.type)
             if field.type is ty.Any:

pydra/design/tests/test_workflow.py

@@ -1,29 +1,37 @@
+from operator import attrgetter
+import pytest
+import attrs
 from pydra.engine.workflow import Workflow
 from pydra.engine.specs import LazyField
 import typing as ty
-from pydra.design import shell, python, workflow, list_fields
+from pydra.design import shell, python, workflow, list_fields, TaskSpec
 from fileformats import video, image
 
 
 def test_workflow():
 
-    @workflow.define
-    def MyTestWorkflow(a, b):
+    # NB: We use PascalCase (i.e. class names) as it is translated into a class
 
-        @python.define
-        def Add(a, b):
-            return a + b
+    @python.define
+    def Add(a, b):
+        return a + b
 
-        @python.define
-        def Mul(a, b):
-            return a * b
+    @python.define
+    def Mul(a, b):
+        return a * b
 
+    @workflow.define
+    def MyTestWorkflow(a, b):
         add = workflow.add(Add(a=a, b=b))
         mul = workflow.add(Mul(a=add.out, b=b))
         return mul.out
 
     constructor = MyTestWorkflow().constructor
     assert constructor.__name__ == "MyTestWorkflow"
+
+    # The constructor function is included as a part of the specification so it is
+    # included in the hash by default and can be overridden if needed. Not 100% sure
+    # if this is a good idea or not
     assert list_fields(MyTestWorkflow) == [
         workflow.arg(name="a"),
         workflow.arg(name="b"),
@@ -36,29 +44,42 @@ def Mul(a, b):
     wf = Workflow.construct(workflow_spec)
     assert wf.inputs.a == 1
     assert wf.inputs.b == 2.0
-    assert wf.outputs.out == LazyField(name="Mul", field="out", type=ty.Any)
+    assert wf.outputs.out == LazyField(
+        name="Mul", field="out", type=ty.Any, type_checked=True
+    )
+
+    # Nodes are named after the specs by default
     assert list(wf.node_names) == ["Add", "Mul"]
 
 
 def test_shell_workflow():
 
     @workflow.define
-    def MyTestShellWorkflow(input_video: video.Mp4, watermark: image.Png) -> video.Mp4:
+    def MyTestShellWorkflow(
+        input_video: video.Mp4,
+        watermark: image.Png,
+        watermark_dims: tuple[int, int] = (10, 10),
+    ) -> video.Mp4:
 
         add_watermark = workflow.add(
             shell.define(
-                "ffmpeg -i <in_video> -i <watermark:image/png> -filter_complex <filter> <out|out_video>"
-            )(in_video=input_video, watermark=watermark, filter="overlay=10:10"),
+                "ffmpeg -i <in_video> -i <watermark:image/png> "
+                "-filter_complex <filter> <out|out_video>"
+            )(
+                in_video=input_video,
+                watermark=watermark,
+                filter="overlay={}:{}".format(*watermark_dims),
+            ),
             name="add_watermark",
         )
         output_video = workflow.add(
             shell.define(
-                (
-                    "HandBrakeCLI -i <in_video> -o <out|out_video> "
-                    "--width <width:int> --height <height:int>"
-                ),
-                # this specifies that this output is required even though it has a flag,
-                # optional inputs and outputs are of type * | None
+                "HandBrakeCLI -i <in_video> -o <out|out_video> "
+                "--width <width:int> --height <height:int>",
+                # By default any input/output specified with a flag (e.g. -i <in_video>)
+                # is considered optional, i.e. of type `FsObject | None`, and therefore
+                # won't be used by default. By overriding this with non-optional types,
+                # the fields are specified as being required.
                 inputs={"in_video": video.Mp4},
                 outputs={"out_video": video.Mp4},
             )(in_video=add_watermark.out_video, width=1280, height=720),
@@ -72,6 +93,7 @@ def MyTestShellWorkflow(input_video: video.Mp4, watermark: image.Png) -> video.M
     assert list_fields(MyTestShellWorkflow) == [
         workflow.arg(name="input_video", type=video.Mp4),
         workflow.arg(name="watermark", type=image.Png),
+        workflow.arg(name="watermark_dims", type=tuple[int, int], default=(10, 10)),
         workflow.arg(name="constructor", type=ty.Callable, default=constructor),
     ]
     assert list_fields(MyTestShellWorkflow.Outputs) == [
@@ -87,12 +109,86 @@ def MyTestShellWorkflow(input_video: video.Mp4, watermark: image.Png) -> video.M
     assert wf.inputs.input_video == input_video
     assert wf.inputs.watermark == watermark
     assert wf.outputs.output_video == LazyField(
-        name="resize", field="out_video", type=video.Mp4
+        name="resize", field="out_video", type=video.Mp4, type_checked=True
     )
     assert list(wf.node_names) == ["add_watermark", "resize"]
 
 
-def test_workflow_alt_syntax():
+def test_workflow_canonical():
+
+    # NB: We use PascalCase (i.e. class names) as it is translated into a class
+
+    @python.define
+    def Add(a, b):
+        return a + b
+
+    @python.define
+    def Mul(a, b):
+        return a * b
+
+    def a_converter(value):
+        if value is attrs.NOTHING:
+            return value
+        return float(value)
+
+    @workflow.define
+    class MyTestWorkflow(TaskSpec["MyTestWorkflow.Outputs"]):
+
+        a: int
+        b: float = workflow.arg(
+            help_string="A float input",
+            converter=a_converter,
+        )
+
+        @staticmethod
+        def constructor(a, b):
+            add = workflow.add(Add(a=a, b=b))
+            mul = workflow.add(Mul(a=add.out, b=b))
+            return mul.out
+
+        class Outputs:
+            out: float
+
+    constructor = MyTestWorkflow().constructor
+    assert constructor.__name__ == "constructor"
+
+    # The constructor function is included as a part of the specification so it is
+    # included in the hash by default and can be overridden if needed. Not 100% sure
+    # if this is a good idea or not
+    assert sorted(list_fields(MyTestWorkflow), key=attrgetter("name")) == [
+        workflow.arg(name="a", type=int),
+        workflow.arg(
+            name="b", type=float, help_string="A float input", converter=a_converter
+        ),
+        workflow.arg(name="constructor", type=ty.Callable, default=constructor),
+    ]
+    assert list_fields(MyTestWorkflow.Outputs) == [
+        workflow.out(name="out", type=float),
+    ]
+    workflow_spec = MyTestWorkflow(a=1, b=2.0)
+    wf = Workflow.construct(workflow_spec)
+    assert wf.inputs.a == 1
+    assert wf.inputs.b == 2.0
+    assert wf.outputs.out == LazyField(
+        name="Mul", field="out", type=ty.Any, type_checked=True
+    )
+
+    # Nodes are named after the specs by default
+    assert list(wf.node_names) == ["Add", "Mul"]
+
+
+def test_direct_access_of_workflow_object():
+
+    @python.define(inputs={"x": float}, outputs={"z": float})
+    def Add(x, y):
+        return x + y
+
+    def Mul(x, y):
+        return x * y
+
+    @python.define(outputs=["divided"])
+    def Divide(x, y):
+        return x / y
 
     @workflow.define(outputs=["out1", "out2"])
     def MyTestWorkflow(a: int, b: float) -> tuple[float, float]:
@@ -107,22 +203,11 @@ def MyTestWorkflow(a: int, b: float) -> tuple[float, float]:
             out2: The second output
         """
 
-        @python.define(inputs={"x": float}, outputs={"out": float})
-        def Add(x, y):
-            return x + y
-
-        def Mul(x, y):
-            return x * y
-
-        @python.define(outputs=["divided"])
-        def Divide(x, y):
-            return x / y
-
         wf = workflow.this()
 
         add = wf.add(Add(x=a, y=b), name="addition")
-        mul = wf.add(python.define(Mul, outputs={"out": float})(x=add.out, y=b))
-        divide = wf.add(Divide(x=wf["addition"].lzout.out, y=mul.out), name="division")
+        mul = wf.add(python.define(Mul, outputs={"out": float})(x=add.z, y=b))
+        divide = wf.add(Divide(x=wf["addition"].lzout.z, y=mul.out), name="division")
 
         # Alter one of the inputs to a node after it has been initialised
         wf["Mul"].inputs.y *= 2
@@ -144,33 +229,38 @@ def Divide(x, y):
     wf = Workflow.construct(workflow_spec)
     assert wf.inputs.a == 1
     assert wf.inputs.b == 2.0
-    assert wf.outputs.out1 == LazyField(name="Mul", field="out", type=float)
-    assert wf.outputs.out2 == LazyField(name="division", field="divided", type=ty.Any)
+    assert wf.outputs.out1 == LazyField(
+        name="Mul", field="out", type=float, type_checked=True
+    )
+    assert wf.outputs.out2 == LazyField(
+        name="division", field="divided", type=ty.Any, type_checked=True
+    )
     assert list(wf.node_names) == ["addition", "Mul", "division"]
 
 
 def test_workflow_set_outputs_directly():
 
-    @workflow.define(outputs={"out1": float, "out2": float})
-    def MyTestWorkflow(a: int, b: float):
+    @python.define
+    def Add(a, b):
+        return a + b
 
-        @python.define
-        def Add(a, b):
-            return a + b
+    @python.define
+    def Mul(a, b):
+        return a * b
 
-        @python.define
-        def Mul(a, b):
-            return a * b
+    @workflow.define(outputs={"out1": float, "out2": float})
+    def MyTestWorkflow(a: int, b: float):
 
         wf = workflow.this()
-
         add = wf.add(Add(a=a, b=b))
         wf.add(Mul(a=add.out, b=b))
 
+        # Set the outputs of the workflow directly instead of returning them them in
+        # a tuple
         wf.outputs.out2 = add.out  # Using the returned lzout outputs
         wf.outputs.out1 = wf["Mul"].lzout.out  # accessing the lzout outputs via getitem
 
-        # no return required when the outputs are set directly
+        # no return is used when the outputs are set directly
 
     assert list_fields(MyTestWorkflow) == [
         workflow.arg(name="a", type=int),
@@ -187,6 +277,63 @@ def Mul(a, b):
     wf = Workflow.construct(workflow_spec)
     assert wf.inputs.a == 1
     assert wf.inputs.b == 2.0
-    assert wf.outputs.out1 == LazyField(name="Mul", field="out", type=ty.Any)
-    assert wf.outputs.out2 == LazyField(name="Add", field="out", type=ty.Any)
+    assert wf.outputs.out1 == LazyField(
+        name="Mul", field="out", type=ty.Any, type_checked=True
+    )
+    assert wf.outputs.out2 == LazyField(
+        name="Add", field="out", type=ty.Any, type_checked=True
+    )
     assert list(wf.node_names) == ["Add", "Mul"]
+
+
+def test_workflow_split_combine():
+
+    @python.define
+    def Mul(x: float, y: float) -> float:
+        return x * y
+
+    @python.define
+    def Sum(x: list[float]) -> float:
+        return sum(x)
+
+    @workflow.define
+    def MyTestWorkflow(a: list[int], b: list[float]) -> list[float]:
+
+        wf = workflow.this()
+        mul = wf.add(Mul())
+        # We could avoid having to specify the splitter and combiner on a separate
+        # line by making 'split' and 'combine' reserved keywords for Outputs class attrs
+        wf["Mul"].split(x=a, y=b).combine("a")
+        sum = wf.add(Sum(x=mul.out))
+        return sum.out
+
+    wf = Workflow.construct(MyTestWorkflow(a=[1, 2, 3], b=[1.0, 10.0, 100.0]))
+    assert wf["Mul"]._state.splitter == ["x", "y"]
+    assert wf["Mul"]._state.combiner == ["x"]
+
+
+def test_workflow_split_after_access_fail():
+    """It isn't possible to split/combine a node after one of its outputs has been type
+    checked as this changes the type of the outputs and renders the type checking
+    invalid
+    """
+
+    @python.define
+    def Add(x, y):
+        return x + y
+
+    @python.define
+    def Mul(x, y):
+        return x * y
+
+    @workflow.define
+    def MyTestWorkflow(a: list[int], b: list[float]) -> list[float]:
+
+        wf = workflow.this()
+        add = wf.add(Add())
+        mul = wf.add(Mul(x=add.out, y=2.0))  # << Add.out is accessed here
+        wf["Add"].split(x=a, y=b).combine("x")
+        return mul.out
+
+    with pytest.raises(RuntimeError, match="Outputs .* have already been accessed"):
+        Workflow.construct(MyTestWorkflow(a=[1, 2, 3], b=[1.0, 10.0, 100.0]))

pydra/design/workflow.py

@@ -1,7 +1,7 @@
 import typing as ty
 import inspect
 import attrs
-from pydra.engine.task import FunctionTask
+from pydra.engine.core import WorkflowTask
 from pydra.engine.workflow import Workflow
 from .base import (
     Arg,
@@ -148,7 +148,7 @@ def make(wrapped: ty.Callable | type) -> TaskSpec:
             parsed_inputs[inpt_name].lazy = True
 
         interface = make_task_spec(
-            FunctionTask,
+            WorkflowTask,
             parsed_inputs,
             parsed_outputs,
             name=name,