added option to explicitly route connections from/to explicit inputs and outputs

Author: tclose

example-specs/pkg-gen/niworkflows.yaml

@@ -5,6 +5,7 @@ niworkflows:
     - niworkflows.interfaces.bids.ReadSidecarJSON
     - niworkflows.interfaces.fixes.FixHeaderApplyTransforms
     - niworkflows.interfaces.fixes.FixN4BiasFieldCorrection
+    - niworkflows.interfaces.fixes.FixHeaderRegistration
     - niworkflows.interfaces.header.SanitizeImage
     - niworkflows.interfaces.images.RobustAverage
     - niworkflows.interfaces.morphology.BinaryDilation

nipype2pydra/package.py

@@ -422,10 +422,10 @@ def collect_intra_pkg_objects(used: UsedSymbols, port_nipype: bool = True):
             intra_pkg_modules[conv.nipype_module_name].add(conv.nipype_object)
             collect_intra_pkg_objects(conv.used_symbols)
 
-        for converter in tqdm(
+        for workflow in tqdm(
             workflows_to_include, "converting workflows from Nipype to Pydra syntax"
         ):
-            all_used = converter.write(
+            all_used = workflow.write(
                 package_root,
                 already_converted=already_converted,
             )

nipype2pydra/pkg_gen/__init__.py

@@ -403,7 +403,7 @@ def generate_callables(self, nipype_interface) -> str:
                 if output_name not in INBUILT_NIPYPE_TRAIT_NAMES:
                     callables_str += (
                         f"def {output_name}_callable(output_dir, inputs, stdout, stderr):\n"
-                        "    parsed_inputs = {}"
+                        "    parsed_inputs = {}\n"
                         "    outputs = _list_outputs(output_dir=output_dir, inputs=inputs, stdout=stdout, stderr=stderr, parsed_inputs=parsed_inputs)\n"
                         '    return outputs["' + output_name + '"]\n\n'
                     )
@@ -422,7 +422,7 @@ def generate_callables(self, nipype_interface) -> str:
                 callables_str, fast=False, mode=black.FileMode()
             )
         except black.parsing.InvalidInput as e:
-            with open(Path("~/Desktop/gen-code.py").expanduser(), "w") as f:
+            with open(Path("~/unparsable-gen-code.py").expanduser(), "w") as f:
                 f.write(callables_str)
             raise RuntimeError(
                 f"Black could not parse generated code: {e}\n\n{callables_str}"
@@ -1026,9 +1026,7 @@ def process_method(
         )
         if hasattr(nipype_interface, "_cmd"):
             body = body.replace("self.cmd", f'"{nipype_interface._cmd}"')
-        body = re.sub(
-            r"getattr\(self\.inputs, (\w+), None\)", r"inputs.get(\1)", body
-        )
+        body = re.sub(r"getattr\(self\.inputs, (\w+), None\)", r"inputs.get(\1)", body)
         body = re.sub(r"getattr\(self\.inputs, (\w+)\)", r"inputs[\1]", body)
         if attrs_as_parsed_inputs:
             body = re.sub(

nipype2pydra/statements/workflow_build.py

@@ -106,7 +106,7 @@ class DynamicField(VarField):
     callable: ty.Callable = attrs.field()
 
     def __repr__(self):
-        return f"DelayedVarField({self.varname}, callable={self.callable})"
+        return f"DynamicField({self.varname}, callable={self.callable})"
 
 
 @attrs.define

nipype2pydra/workflow.py

@@ -36,6 +36,7 @@
     WorkflowInitStatement,
     AssignmentStatement,
     OtherStatement,
+    DynamicField,
 )
 import nipype2pydra.package
 
@@ -94,8 +95,7 @@ class WorkflowInterfaceField:
         factory=list,
         metadata={
             "help": (
-                "node-name/field-name pairs of other fields that are to be routed to "
-                "from other node fields to this input/output",
+                "node-name/field-name pairs of additional fields that this input/output replaces",
             )
         },
     )
@@ -159,6 +159,11 @@ def __hash__(self):
 @attrs.define
 class WorkflowInput(WorkflowInterfaceField):
 
+    connections: ty.Tuple[ty.Tuple[str, str]] = attrs.field(
+        converter=lambda lst: tuple(sorted(tuple(t) for t in lst)),
+        factory=list,
+        metadata={"help": ("Explicit connections to be made from this input field",)},
+    )
     out_conns: ty.List[ConnectionStatement] = attrs.field(
         factory=list,
         eq=False,
@@ -170,9 +175,7 @@ class WorkflowInput(WorkflowInterfaceField):
             )
         },
     )
-
     include: bool = attrs.field(
-        default=False,
         eq=False,
         hash=False,
         metadata={
@@ -183,13 +186,22 @@ class WorkflowInput(WorkflowInterfaceField):
         },
     )
 
+    @include.default
+    def _include_default(self) -> bool:
+        return bool(self.connections)
+
     def __hash__(self):
         return super().__hash__()
 
 
 @attrs.define
 class WorkflowOutput(WorkflowInterfaceField):
 
+    connection: ty.Tuple[str, str] = attrs.field(
+        converter=tuple,
+        factory=list,
+        metadata={"help": ("Explicit connection to be made to this output field",)},
+    )
     in_conns: ty.List[ConnectionStatement] = attrs.field(
         factory=list,
         eq=False,
@@ -413,6 +425,12 @@ def get_input_from_conn(self, conn: ConnectionStatement) -> WorkflowInput:
         """
         Returns the name of the input field in the workflow for the given node and field
         escaped by the prefix of the node if present"""
+        if isinstance(conn.source_out, DynamicField):
+            logger.warning(
+                f"Not able to connect inputs from {conn.source_name}:{conn.source_out}->"
+                f"{conn.target_name}:{conn.target_in} properly due to adynamic-field "
+                "just connecting to source input for now"
+            )
         try:
             return self.make_input(
                 field_name=conn.source_out,
@@ -1036,7 +1054,7 @@ def prepare_connections(self):
                 # append to parsed statements so set_output can be set
                 self.parsed_statements.append(conn_stmt)
         while self._unprocessed_connections:
-            conn = self._unprocessed_connections.pop()
+            conn = self._unprocessed_connections.pop(0)
             try:
                 inpt = self.get_input_from_conn(conn)
             except KeyError:
@@ -1056,6 +1074,47 @@ def prepare_connections(self):
                 conn.target_in = outpt.name
                 outpt.in_conns.append(conn)
 
+        # Overwrite connections with explict connections
+        for inpt in list(self.inputs.values()):
+            for target_name, target_in in inpt.connections:
+                conn = ConnectionStatement(
+                    indent="    ",
+                    source_name=None,
+                    source_out=inpt.name,
+                    target_name=target_name,
+                    target_in=target_in,
+                    workflow_converter=self,
+                )
+                for tgt_node in self.nodes[conn.target_name]:
+                    try:
+                        existing_conn = next(
+                            c for c in tgt_node.in_conns if c.target_in == target_in
+                        )
+                    except StopIteration:
+                        pass
+                    else:
+                        tgt_node.in_conns.remove(existing_conn)
+                        self.inputs[existing_conn.source_out].out_conns.remove(
+                            existing_conn
+                        )
+                    inpt.out_conns.append(conn)
+                    tgt_node.add_input_connection(conn)
+
+        for outpt in list(self.outputs.values()):
+            if outpt.connection:
+                source_name, source_out = outpt.connection
+                conn = ConnectionStatement(
+                    indent="    ",
+                    source_name=source_name,
+                    source_out=source_out,
+                    target_name=None,
+                    target_in=outpt.name,
+                    workflow_converter=self,
+                )
+                for src_node in self.nodes[conn.source_name]:
+                    src_node.add_output_connection(conn)
+                    outpt.in_conns.append(conn)
+
     def _parse_statements(self, func_body: str) -> ty.Tuple[
         ty.List[
             ty.Union[