debugging shelltask

Author: tclose

pydra/design/base.py

@@ -350,6 +350,8 @@ def get_fields(klass, field_type, auto_attribs, helps) -> dict[str, Field]:
                 if not atr.help:
                     atr.help = helps.get(atr_name, "")
             elif atr_name in type_hints:
+                if atr_name.startswith("_"):
+                    continue
                 if atr_name in fields_dict:
                     fields_dict[atr_name].type = type_hints[atr_name]
                 elif auto_attribs:
@@ -361,6 +363,8 @@ def get_fields(klass, field_type, auto_attribs, helps) -> dict[str, Field]:
                     )
         if auto_attribs:
             for atr_name, type_ in type_hints.items():
+                if atr_name.startswith("_"):
+                    continue
                 if atr_name not in list(fields_dict) + ["Task", "Outputs"]:
                     fields_dict[atr_name] = field_type(
                         name=atr_name, type=type_, help=helps.get(atr_name, "")

pydra/design/shell.py

@@ -28,7 +28,6 @@
     MultiInputObj,
     TypeParser,
     is_optional,
-    optional_type,
 )
 
 if ty.TYPE_CHECKING:
@@ -85,7 +84,7 @@ class arg(Arg):
         If nothing is provided the field will be inserted between all fields with
         nonnegative positions and fields with negative positions.
     sep: str, optional
-        A separator if a list is provided as a value.
+        A separator if a sequence type is provided as a value, by default " ".
     container_path: bool, optional
         If True a path will be consider as a path inside the container (and not as a
         local path, by default it is False
@@ -99,45 +98,11 @@ class arg(Arg):
 
     argstr: str | None = ""
     position: int | None = None
-    sep: str | None = attrs.field()
+    sep: str = " "
     allowed_values: list | None = None
     container_path: bool = False  # IS THIS STILL USED??
     formatter: ty.Callable | None = None
 
-    @sep.default
-    def _sep_default(self):
-        return " " if self.type is tuple or ty.get_origin(self.type) is tuple else None
-
-    @sep.validator
-    def _validate_sep(self, _, sep):
-        if self.type is MultiInputObj:
-            tp = ty.Any
-        elif ty.get_origin(self.type) is MultiInputObj:
-            tp = ty.get_args(self.type)[0]
-        else:
-            tp = self.type
-        if is_optional(tp):
-            tp = optional_type(tp)
-        if tp is ty.Any:
-            return
-        origin = ty.get_origin(tp) or tp
-
-        if (
-            inspect.isclass(origin)
-            and issubclass(origin, ty.Sequence)
-            and tp is not str
-        ):
-            if sep is None and not self.readonly:
-                raise ValueError(
-                    f"A value to 'sep' must be provided when type is iterable {tp} "
-                    f"for field {self.name!r}"
-                )
-        elif sep is not None:
-            raise ValueError(
-                f"sep ({sep!r}) can only be provided when type is iterable {tp} "
-                f"for field {self.name!r}"
-            )
-
 
 @attrs.define(kw_only=True)
 class out(Out):

pydra/engine/lazy.py

@@ -152,7 +152,7 @@ def _get_value(
             the resolved value of the lazy-field
         """
 
-        jobs = graph.node(self._node.name).matching_jobs(state_index)
+        jobs = graph.node(self._node.name).get_jobs(state_index)
 
         def retrieve_from_job(job: "Task[DefType]") -> ty.Any:
             if job.errored:
@@ -184,7 +184,9 @@ def retrieve_from_job(job: "Task[DefType]") -> ty.Any:
             val = self._apply_cast(val)
             return val
 
-        if not self._node.state or not self._node.state.depth(before_combine=True):
+        if not isinstance(jobs, StateArray):
+            return retrieve_from_job(jobs)
+        elif not self._node.state or not self._node.state.depth(before_combine=True):
             assert len(jobs) == 1
             return retrieve_from_job(jobs[0])
         elif not self._node.state.keys_final:  # all states are combined over

pydra/engine/specs.py

@@ -1089,6 +1089,7 @@ def _command_args(self, values: dict[str, ty.Any]) -> list[str]:
         self._check_resolved()
         self._check_rules()
         # Drop none/empty values and optional path fields that are set to false
+        values = copy(values)  # Create a copy so we can drop items from the dictionary
         for field in list_fields(self):
             fld_value = values[field.name]
             if fld_value is None or (is_multi_input(field.type) and fld_value == []):
@@ -1185,7 +1186,7 @@ def _command_pos_args(
             call_args_val = {}
             for argnm in call_args.args:
                 if argnm == "field":
-                    call_args_val[argnm] = value
+                    call_args_val[argnm] = field
                 elif argnm == "inputs":
                     call_args_val[argnm] = values
                 else:
@@ -1223,31 +1224,31 @@ def _format_arg(self, field: shell.arg, values: dict[str, ty.Any]) -> list[str]:
             and isinstance(value, ty.Iterable)
             and not isinstance(value, (str, bytes))
         ):
-            field.argstr = field.argstr.replace("...", "")
+            argstr = field.argstr.replace("...", "")
             # if argstr has a more complex form, with "{input_field}"
-            if "{" in field.argstr and "}" in field.argstr:
+            if "{" in argstr and "}" in argstr:
                 argstr_formatted_l = []
                 for val in value:
                     split_values = copy(values)
                     split_values[field.name] = val
-                    argstr_f = argstr_formatting(field.argstr, split_values)
+                    argstr_f = argstr_formatting(argstr, split_values)
                     argstr_formatted_l.append(f" {argstr_f}")
-                cmd_el_str = field.sep.join(argstr_formatted_l)
+                cmd_el_str = " ".join(argstr_formatted_l)
             else:  # argstr has a simple form, e.g. "-f", or "--f"
-                cmd_el_str = field.sep.join([f" {field.argstr} {val}" for val in value])
+                cmd_el_str = " ".join([f" {argstr} {val}" for val in value])
         else:
             # in case there are ... when input is not a list
-            field.argstr = field.argstr.replace("...", "")
+            argstr = field.argstr.replace("...", "")
             if isinstance(value, ty.Iterable) and not isinstance(value, (str, bytes)):
                 cmd_el_str = field.sep.join([str(val) for val in value])
                 value = cmd_el_str
             # if argstr has a more complex form, with "{input_field}"
-            if "{" in field.argstr and "}" in field.argstr:
-                cmd_el_str = field.argstr.replace(f"{{{field.name}}}", str(value))
+            if "{" in argstr and "}" in argstr:
+                cmd_el_str = argstr.replace(f"{{{field.name}}}", str(value))
                 cmd_el_str = argstr_formatting(cmd_el_str, values)
             else:  # argstr has a simple form, e.g. "-f", or "--f"
                 if value:
-                    cmd_el_str = f"{field.argstr} {value}"
+                    cmd_el_str = f"{argstr} {value}"
                 else:
                     cmd_el_str = ""
         return split_cmd(cmd_el_str)

pydra/engine/state.py

@@ -879,9 +879,6 @@ def prepare_states(
         if self.other_states:
             st: State
             for nm, (st, _) in self.other_states.items():
-                # I think now this if is never used
-                if not hasattr(st, "states_ind"):
-                    st.prepare_states(self.inputs, cont_dim=cont_dim)
                 self.inputs.update(st.inputs)
                 self.cont_dim.update(st.cont_dim)
 

pydra/engine/submitter.py

@@ -538,7 +538,7 @@ def __init__(
         self.unrunnable = defaultdict(list)
         # Prepare the state to be run
         if node.state:
-            self.state = deepcopy(node.state)
+            self.state = node.state
             self.state.prepare_states(self.node.state_values)
             self.state.prepare_inputs()
         else:
@@ -564,34 +564,43 @@ def _definition(self) -> "Node":
         return self.node._definition
 
     @property
-    def tasks(self) -> ty.Iterable["Task[DefType]"]:
+    def tasks(self) -> ty.Generator["Task[DefType]", None, None]:
         if self._tasks is None:
             self._tasks = {t.state_index: t for t in self._generate_tasks()}
         return self._tasks.values()
 
-    def matching_jobs(self, index: int | None = None) -> "StateArray[Task]":
+    def get_jobs(
+        self, index: int | None = None, as_array: bool = False
+    ) -> "Task | StateArray[Task]":
         """Get the jobs that match a given state index.
 
         Parameters
         ----------
         index : int, optional
             The index of the state of the task to get, by default None
+        as_array : bool, optional
+            Whether to return the tasks in a state-array object, by default if the index
+            matches
 
         Returns
         -------
-        matching : StateArray[Task]
-            The tasks that match the given index
+        matching : Task | StateArray[Task]
+            The task or tasks that match the given index
         """
         matching = StateArray()
         if self.tasks:
             try:
-                matching.append(self._tasks[index])
+                task = self._tasks[index]
             except KeyError:
+                if index is None:
+                    return StateArray(self._tasks.values())
                 # Select matching tasks and return them in nested state-array objects
                 for ind, task in self._tasks.items():
-                    if ind.matches(index):
-                        matching.append(task)
-
+                    matching.append(task)
+            else:
+                if not as_array:
+                    return task
+                matching.append(task)
         return matching
 
     @property
@@ -657,7 +666,7 @@ def _generate_tasks(self) -> ty.Iterable["Task[DefType]"]:
                 name=self.node.name,
             )
         else:
-            for index, split_defn in self._split_definition().items():
+            for index, split_defn in enumerate(self._split_definition()):
                 yield Task(
                     definition=split_defn,
                     submitter=self.submitter,
@@ -707,7 +716,7 @@ def _split_definition(self) -> dict[int, "TaskDef[OutputType]"]:
         if not self.node.state:
             return {None: self.node._definition}
         split_defs = []
-        for i, input_ind in enumerate(self.node.state.inputs_ind):
+        for input_ind in self.node.state.inputs_ind:
             resolved = {}
             for inpt_name in set(self.node.input_names):
                 value = getattr(self._definition, inpt_name)
@@ -716,13 +725,13 @@ def _split_definition(self) -> dict[int, "TaskDef[OutputType]"]:
                     resolved[inpt_name] = value._get_value(
                         workflow=self.workflow,
                         graph=self.graph,
-                        state_index=i,
+                        state_index=input_ind[state_key],
                     )
                 elif state_key in input_ind:
                     resolved[inpt_name] = self.node.state._get_element(
                         value=value,
                         field_name=inpt_name,
-                        ind=i,
+                        ind=input_ind[state_key],
                     )
             split_defs.append(attrs.evolve(self.node._definition, **resolved))
         return split_defs
@@ -754,7 +763,8 @@ def get_runnable_tasks(self, graph: DiGraph) -> list["Task[DefType]"]:
             pred: NodeExecution
             is_runnable = True
             for pred in graph.predecessors[self.node.name]:
-                pred_inds = [j.state_index for j in pred.matching_jobs(index)]
+                pred_jobs: StateArray[Task] = pred.get_jobs(index, as_array=True)
+                pred_inds = [j.state_index for j in pred_jobs]
                 if not all(i in pred.successful for i in pred_inds):
                     is_runnable = False
                     blocked = True