relaxed task definition validation to accept use cases found in ANTs, FSL, Freesurfer, etc... task packages

Author: tclose

pydra/compose/base/__init__.py

@@ -6,6 +6,7 @@
     check_explicit_fields_are_none,
     extract_fields_from_class,
     is_set,
+    sanitize_xor,
 )
 from .task import Task, Outputs
 from .builder import build_task_class

pydra/compose/base/builder.py

@@ -19,6 +19,7 @@
     is_lazy,
 )
 from .field import Field, Arg, Out
+from .helpers import sanitize_xor
 
 
 def build_task_class(
@@ -65,15 +66,7 @@ def build_task_class(
     klass : type
         The class created using the attrs package
     """
-
-    # Convert a single xor set into a set of xor sets
-    if not xor:
-        xor = frozenset()
-    elif all(isinstance(x, str) or x is None for x in xor):
-        xor = frozenset([frozenset(xor)])
-    else:
-        xor = frozenset(frozenset(x) for x in xor)
-
+    xor = sanitize_xor(xor)
     spec_type._check_arg_refs(inputs, outputs, xor)
 
     # Check that the field attributes are valid after all fields have been set

pydra/compose/base/field.py

@@ -5,7 +5,7 @@
 from attrs.converters import default_if_none
 from fileformats.core import to_mime
 from fileformats.generic import File, FileSet
-from pydra.utils.typing import TypeParser, is_optional, is_type, is_union
+from pydra.utils.typing import TypeParser, is_optional, is_truthy_falsy, is_type, is_union
 from pydra.utils.general import get_fields, wrap_text
 import attrs
 
@@ -229,10 +229,10 @@ def mandatory(self):
 
     @requires.validator
     def _requires_validator(self, _, value):
-        if value and self.type not in (ty.Any, bool) and not is_optional(self.type):
+        if value and not is_truthy_falsy(self.type):
             raise ValueError(
-                f"Fields with requirements must be of optional type (i.e. in union "
-                f"with None) or boolean, not type {self.type} ({self!r})"
+                f"Fields with requirements must be of optional (i.e. in union "
+                f"with None) or truthy/falsy type, not type {self.type} ({self!r})"
             )
 
     def markdown_listing(

pydra/compose/base/helpers.py

@@ -379,6 +379,20 @@ def check_explicit_fields_are_none(klass, inputs, outputs):
         )
 
 
+def sanitize_xor(
+    xor: ty.Sequence[str | None] | ty.Sequence[ty.Sequence[str | None]],
+) -> set[frozenset[str]]:
+    """Convert a list of xor sets into a set of frozensets"""
+    # Convert a single xor set into a set of xor sets
+    if not xor:
+        xor = frozenset()
+    elif all(isinstance(x, str) or x is None for x in xor):
+        xor = frozenset([frozenset(xor)])
+    else:
+        xor = frozenset(frozenset(x) for x in xor)
+    return xor
+
+
 def extract_fields_from_class(
     spec_type: type["Task"],
     outputs_type: type["Outputs"],

pydra/compose/base/task.py

@@ -4,7 +4,7 @@
 from copy import copy
 from typing import Self
 import attrs.validators
-from pydra.utils.typing import is_optional, is_fileset_or_union
+from pydra.utils.typing import is_optional, is_fileset_or_union, is_truthy_falsy
 from pydra.utils.general import get_fields
 from pydra.utils.typing import StateArray, is_lazy
 from pydra.utils.hash import hash_function
@@ -595,17 +595,17 @@ def _check_arg_refs(
         for xor_set in xor:
             if unrecognised := xor_set - (input_names | {None}):
                 raise ValueError(
-                    f"'Unrecognised' field names in referenced in the xor {xor_set} "
+                    f"Unrecognised field names in referenced in the xor {xor_set}: "
                     + str(list(unrecognised))
                 )
             for field_name in xor_set:
                 if field_name is None:  # i.e. none of the fields being set is valid
                     continue
                 type_ = inputs[field_name].type
-                if type_ not in (ty.Any, bool) and not is_optional(type_):
+                if not is_truthy_falsy(type_):
                     raise ValueError(
-                        f"Fields included in a 'xor' ({field_name!r}) must be of boolean "
-                        f"or optional types, not type {type_}"
+                        f"Fields included in a 'xor' ({field_name!r}) must be an optional type or a"
+                        f"truthy/falsy type, not type {type_}"
                     )
 
     def _check_resolved(self):

pydra/compose/shell/builder.py

@@ -21,6 +21,7 @@
     extract_fields_from_class,
     ensure_field_objects,
     build_task_class,
+    sanitize_xor,
     NO_DEFAULT,
 )
 from pydra.utils.typing import (
@@ -208,7 +209,7 @@ def make(
         )
 
         # Set positions for the remaining inputs that don't have an explicit position
-        position_stack = remaining_positions(list(parsed_inputs.values()))
+        position_stack = remaining_positions(list(parsed_inputs.values()), xor=xor)
         for inpt in parsed_inputs.values():
             if inpt.name == "append_args":
                 continue
@@ -526,7 +527,10 @@ def from_type_str(type_str) -> type:
 
 
 def remaining_positions(
-    args: list[Arg], num_args: int | None = None, start: int = 0
+    args: list[Arg],
+    num_args: int | None = None,
+    start: int = 0,
+    xor: set[frozenset[str]] | None = None,
 ) -> ty.List[int]:
     """Get the remaining positions for input fields
 
@@ -536,6 +540,10 @@ def remaining_positions(
         The list of input fields
     num_args : int, optional
         The number of arguments, by default it is the length of the args
+    start : int, optional
+        The starting position, by default 0
+    xor : set[frozenset[str]], optional
+        A set of mutually exclusive fields, by default None
 
     Returns
     -------
@@ -547,6 +555,7 @@ def remaining_positions(
     ValueError
         If multiple fields have the same position
     """
+    xor = sanitize_xor(xor)
     if num_args is None:
         num_args = len(args) - 1  # Subtract 1 for the 'append_args' field
     # Check for multiple positions
@@ -562,7 +571,7 @@ def remaining_positions(
     if multiple_positions := {
         k: [f"{a.name}({a.position})" for a in v]
         for k, v in positions.items()
-        if len(v) > 1
+        if len(v) > 1 and frozenset(a.name for a in v) not in xor
     }:
         raise ValueError(
             f"Multiple fields have the overlapping positions: {multiple_positions}"

pydra/utils/typing.py

@@ -1070,6 +1070,24 @@ def is_optional(type_: type) -> bool:
     return False
 
 
+def is_container(type_: type) -> bool:
+    """Check if the type is a container, i.e. a list, tuple, or MultiOutputObj"""
+    origin = ty.get_origin(type_)
+    tp = origin if origin else type_
+    return inspect.isclass(tp) and issubclass(tp, ty.Container)
+
+
+def is_truthy_falsy(type_: type) -> bool:
+    """Check if the type is a truthy type, i.e. not None, bool, or typing.Any"""
+    return (
+        type_ in (ty.Any, bool, int, str)
+        or is_optional(type_)
+        or is_container(type_)
+        or hasattr(type_, "__bool__")
+        or hasattr(type_, "__len__")
+    )
+
+
 def optional_type(type_: type) -> type:
     """Gets the non-None args of an optional type (i.e. a union with a None arg)"""
     if is_optional(type_):