debugging task unittests

Author: tclose

nipype2pydra/cli.py

@@ -20,32 +20,45 @@ def cli():
 manually specified aspects of the conversion see
 https://github.com/nipype/nipype2pydra/tree/main/example-specs for examples
 
-OUTPUT_BASE_DIR is the path of generated module file
+PACKAGE_ROOT is the path to the root directory of the package in which to generate the
+converted module file
 """
 )
 @click.argument("yaml-spec", type=click.File())
-@click.argument("output-file", type=Path)
+@click.argument("package-root", type=Path)
 @click.option(
     "-c",
     "--callables",
     type=click.File(),
     default=None,
     help="a Python file containing callable functions required in the command interface",
 )
-def task(yaml_spec, output_file, callables):
+@click.option(
+    "--output-module" "-m",
+    type=str,
+    default=None,
+    help=(
+        "the output module to store the converted task into relative to the `pydra.tasks` "
+        "package. If not provided, then the path relative to `nipype.interfaces` in the "
+        "source interface will be used instead"
+    ),
+)
+def task(yaml_spec, package_root, callables, output_module):
 
     spec = yaml.safe_load(yaml_spec)
 
-    converter = TaskConverter(callables_module=callables, **spec)
-    converter.generate(output_file)
+    converter = TaskConverter(
+        output_module=output_module, callables_module=callables, **spec
+    )
+    converter.generate(package_root)
 
 
 @cli.command(help="Port Nipype workflow creation functions to Pydra")
 @click.argument("yaml-spec", type=click.File())
-@click.argument("output-module-file", type=click.File(mode="w"))
-def workflow(yaml_spec, output_module_file):
+@click.argument("package-root", type=click.File(mode="w"))
+def workflow(yaml_spec, package_root):
 
     spec = yaml.safe_load(yaml_spec)
 
     converter = WorkflowConverter(spec)
-    converter.generate(output_module_file)
+    converter.generate(package_root)

nipype2pydra/task.py

@@ -26,6 +26,7 @@ class TaskConverter:
     doctest: dict = attrs.field(factory=dict)
     tests_inputs: list = attrs.field(factory=list)
     tests_outputs: list = attrs.field(factory=list)
+    output_module: str = attrs.field(default=None)
     callables_module: ModuleType = attrs.field(
         converter=import_module_from_path, default=None
     )
@@ -35,6 +36,18 @@ def output_callables_validator(self, _, output_callables: dict):
         if not output_callables.keys().isdisjoint(self.output_templates.keys()):
             raise Exception("output_callables and output_templates have the same keys")
 
+    def __attrs_post_init__(self):
+        if self.output_module is None:
+            if self.nipype_module.__name__.startswith("nipype.interfaces."):
+                self.output_module = self.nipype_module.__name__[len("nipype.interfaces."):]
+            else:
+                raise RuntimeError(
+                    "Output-module needs to be explicitly provided to task converter "
+                    "when converting Nipype interefaces in non standard locations such "
+                    f"as {self.nipype_module.__name__}.{self.task_name} (i.e. not in "
+                    "nipype.interfaces)"
+                )
+
     @property
     def nipype_interface(self) -> nipype.interfaces.base.BaseInterface:
         return getattr(self.nipype_module, self.task_name)
@@ -47,20 +60,21 @@ def nipype_input_spec(self) -> nipype.interfaces.base.BaseInterfaceInputSpec:
     def nipype_output_spec(self) -> nipype.interfaces.base.BaseTraitedSpec:
         return self.nipype_interface.output_spec()
 
-    def generate(self, output_file: Path):
+    def generate(self, package_root: Path):
         """creating pydra input/output spec from nipype specs
         if write is True, a pydra Task class will be written to the file together with tests
         """
         input_fields, inp_templates = self.convert_input_fields()
         output_fields = self.convert_output_spec(fields_from_template=inp_templates)
 
-        testdir = output_file.parent / "tests"
-        testdir.mkdir()
-        filename_test = testdir / f"test_spec_{output_file.name}"
-        filename_test_run = testdir / f"test_run_{output_file.name}"
-
+        module_path = (Path(package_root) / "pydra" / "tasks").joinpath(self.output_module.split("."))
+        output_file = (module_path / self.task_name.lower()).with_suffix(".py")
         self.write_task(output_file, input_fields, output_fields)
 
+        testdir = module_path / "tests"
+        testdir.mkdir()
+        filename_test = testdir / f"test_spec_{self.task_name.lower()}"
+        filename_test_run = testdir / f"test_run_{self.task_name.lower()}"
         self.write_test(filename_test=filename_test)
         self.write_test(filename_test=filename_test_run, run=True)
 
@@ -131,7 +145,7 @@ def pydra_fld_input(self, field, nm):
                     tp_pdr = str
             elif nm not in self.output_callables:
                 raise Exception(
-                    f"the filed {nm} has genfile=True, but no output template or callables provided"
+                    f"the filed {nm} has genfile=True, but no output template or callables_module provided"
                 )
 
         metadata_pdr.update(metadata_extra_spec)
@@ -346,12 +360,12 @@ def write_test(self, filename_test, run=False):
 
         spec_str = "import os, pytest \nfrom pathlib import Path\n"
         spec_str += (
-            f"from ..{self.task_name.lower()} import {self.task_name} \n\n"
+            f"from pydra.tasks.{self.output_module}.{self.task_name.lower()} import {self.task_name} \n\n"
         )
         if run:
             pass
         spec_str += f"@pytest.mark.parametrize('inputs, outputs', {tests_inp_outp})\n"
-        spec_str += f"def test_{self.task_name}(test_data, inputs, outputs):\n"
+        spec_str += f"def test_{self.task_name.lower()}(test_data, inputs, outputs):\n"
         spec_str += "    in_file = Path(test_data) / 'test.nii.gz'\n"
         spec_str += "    if inputs is None: inputs = {{}}\n"
         spec_str += "    for key, val in inputs.items():\n"

nipype2pydra/workflow.py

@@ -7,7 +7,6 @@
 import black
 from nipype.interfaces.base import isdefined
 from .utils import load_class_or_func
-from nipype.pipeline.engine.workflows import Workflow
 
 
 class WorkflowConverter:
@@ -41,8 +40,7 @@ def node_connections(self, workflow, functions: dict[str, dict], wf_inputs: dict
             connections.update(self.node_connections(nested_wf, functions=functions))
         return connections
 
-
-    def generate(self, format_with_black=False):
+    def generate(self, package_root: str, format_with_black: bool = False):
 
         functions = defaultdict(dict)
         connections = self.node_connections(self.wf, functions=functions)

tests/test_task.py

@@ -11,7 +11,6 @@
     "trait_added",
     "trait_modified",
     "environ",
-    "environ_items",
 ]