implementing the automatic generation of pydra-packages with conversion specs configured

Author: tclose

.flake8

@@ -0,0 +1,17 @@
+
+[flake8]
+doctests = True
+exclude =
+    **/__init__.py
+    *build/
+    docs/sphinxext/
+    docs/tools/
+    docs/conf.py
+    arcana/_version.py
+    versioneer.py
+    docs/source/conf.py
+max-line-length = 88
+select = C,E,F,W,B,B950
+extend-ignore = E203,E501,E129,W503
+per-file-ignores =
+    setup.py:F401

nipype2pydra/task.py

@@ -20,7 +20,7 @@ class TaskConverter:
     nipype_module: ModuleType = attrs.field(converter=import_module_from_path)
     output_requirements: dict = attrs.field(factory=dict)
     inputs_metadata: dict = attrs.field(factory=dict)
-    inputs_drop: dict = attrs.field(factory=dict)
+    inputs_drop: dict = attrs.field(factory=list)
     output_templates: dict = attrs.field(factory=dict)
     output_callables: dict = attrs.field(factory=dict)
     doctest: dict = attrs.field(factory=dict)

nipype2pydra/utils.py

@@ -1,4 +1,5 @@
 import traceback
+import typing as ty
 from types import ModuleType
 import sys
 import os
@@ -18,7 +19,7 @@ def show_cli_trace(result):
     return "".join(traceback.format_exception(*result.exc_info))
 
 
-def import_module_from_path(module_path: Path) -> ModuleType:
+def import_module_from_path(module_path: ty.Union[ModuleType, Path, str]) -> ModuleType:
     if isinstance(module_path, ModuleType) or module_path is None:
         return module_path
     module_path = Path(module_path).resolve()

scripts/pkg_gen/create_packages.py

@@ -0,0 +1,148 @@
+import os
+import typing as ty
+import tempfile
+import re
+import subprocess as sp
+import shutil
+import tarfile
+from pathlib import Path
+import requests
+import click
+import yaml
+
+RESOURCES_DIR = Path(__file__).parent / "resources"
+
+
+def download_tasks_template(output_path: Path):
+    """Downloads the latest pydra-tasks-template to the output path"""
+
+    release_url = (
+        "https://api.github.com/repos/nipype/pydra-tasks-template/releases/latest"
+    )
+    headers = {"Accept": "application/vnd.github.v3+json", "User-Agent": "nipype2pydra"}
+
+    response = requests.get(release_url, headers=headers)
+    if response.status_code != 200:
+        raise RuntimeError(
+            f"Did not find release at '{release_url}'"
+        )
+    data = response.json()
+    tarball_url = data["tarball_url"]
+
+    response = requests.get(tarball_url)
+
+    if response.status_code == 200:
+        # Save the response content to a file
+        with open(output_path, "wb") as f:
+            f.write(response.content)
+    else:
+        raise RuntimeError(
+            f"Could not download the pydra-tasks template at {release_url}"
+        )
+
+
+@click.command(help="Generates stub pydra packages for all nipype interfaces to import")
+@click.argument("output_dir", type=click.Path(path_type=Path))
+@click.option("--work-dir", type=click.Path(path_type=Path), default=None)
+@click.option("--task-template", type=click.Path(path_type=Path), default=None)
+def generate_packages(
+    output_dir: Path, work_dir: ty.Optional[Path], task_template: ty.Optional[Path]
+):
+
+    if work_dir is None:
+        work_dir = Path(tempfile.mkdtemp())
+
+    if task_template is None:
+        task_template_tar = work_dir / "task-template.tar.gz"
+        download_tasks_template(task_template_tar)
+        extract_dir = work_dir / "task_template"
+        with tarfile.open(task_template_tar, 'r:gz') as tar:
+            tar.extractall(path=extract_dir)
+        task_template = extract_dir / next(extract_dir.iterdir())
+
+    with open(Path(__file__).parent.parent.parent / "nipype-interfaces-to-import.yaml") as f:
+        to_import = yaml.load(f, Loader=yaml.SafeLoader)
+
+    # Wipe output dir
+    if output_dir.exists():
+        shutil.rmtree(output_dir)
+    output_dir.mkdir()
+
+    for pkg in to_import["packages"]:
+
+        pkg_dir = output_dir / f"pydra-{pkg}"
+        pkg_dir.mkdir()
+
+        def copy_ignore(_, names):
+            return [n for n in names if n in (".git", "__pycache__", ".pytest_cache")]
+
+        shutil.copytree(task_template, pkg_dir, ignore=copy_ignore)
+
+        auto_conv_dir = pkg_dir / "nipype-auto-conv"
+        specs_dir = auto_conv_dir / "specs"
+        specs_dir.mkdir(parents=True)
+        shutil.copy(RESOURCES_DIR / "nipype-auto-convert.py", auto_conv_dir / "generate")
+        os.chmod(auto_conv_dir / "generate", 0o755)  # make executable
+
+        gh_workflows_dir = pkg_dir / ".github" / "workflows"
+        gh_workflows_dir.mkdir(parents=True)
+        shutil.copy(RESOURCES_DIR / "pythonpackage.yaml", gh_workflows_dir / "pythonpackage.yaml")
+
+        # Add "pydra.tasks.<pkg>.auto to gitignore"
+        with open(pkg_dir / ".gitignore", "a") as f:
+            f.write("\npydra/tasks/{pkg}/auto")
+
+        # rename tasks directory
+        (pkg_dir / "pydra" / "tasks" / "CHANGEME").rename(pkg_dir / "pydra" / "tasks" / pkg)
+
+        # Replace "CHANGEME" string with pkg name
+        for fspath in pkg_dir.glob("**/*"):
+            if fspath.is_dir():
+                continue
+            with open(fspath) as f:
+                contents = f.read()
+            contents = re.sub(r"(?<![0-9a-zA-Z])CHANGEME(?![0-9a-zA-Z])", pkg, contents)
+            with open(fspath, "w") as f:
+                f.write(contents)
+
+        for module, interfaces in to_import["interfaces"].items():
+            if module.split("/")[0] != pkg:
+                continue
+            module_spec_dir = specs_dir.joinpath(*module.split("/"))
+            module_spec_dir.mkdir(parents=True)
+            for interface in interfaces:
+                callables_fspath = module_spec_dir / f"{interface}_callables.py"
+                spec_stub = {
+                    "task_name": interface,
+                    "nipype_module": "nipype.interfaces." + ".".join(module.split("/")),
+                    "output_requirements": "# dict[output-field, list[input-field]] : the required input fields for output-field",
+                    "inputs_metadata": "# dict[input-field, dict[str, Any]] : additional metadata to be inserted into input field",
+                    "inputs_drop": "# list[input-field] : input fields to drop from the spec",
+                    "output_templates": "# dict[input-field, str] : \"output_file_template\" to provide to input field",
+                    "output_callables": f"# dict[output-field, str] : name of function defined in {callables_fspath.name} that retrieves value for output",
+                    "doctest": "# dict[str, Any]: key-value pairs to provide as inputs to the doctest + the expected value of \"cmdline\" as special key-value pair",
+                    "tests_inputs": "# List of inputs to pass to tests",
+                    "tests_outputs": "# list of outputs expected from tests",
+                }
+                yaml_str = yaml.dump(spec_stub, indent=2, sort_keys=False)
+                # strip inserted line-breaks in long strings (so they can be converted to in-line comments)
+                yaml_str = re.sub(r"\n  ", " ", yaml_str)
+                # extract comments after they have been dumped as strings
+                yaml_str = re.sub(r"'#(.*)'", r" # \1", yaml_str)
+                with open(module_spec_dir / (interface + ".yaml"), "w") as f:
+                    f.write(yaml_str)
+                with open(callables_fspath, "w") as f:
+                    f.write(
+                        f'"""Module to put any functions that are referred to in {interface}.yaml"""\n'
+                    )
+
+        sp.check_call("git init", shell=True, cwd=pkg_dir)
+        sp.check_call("git add --all", shell=True, cwd=pkg_dir)
+        sp.check_call('git commit -m"initial commit of generated stubs"', shell=True, cwd=pkg_dir)
+        sp.check_call("git tag 0.1.0", shell=True, cwd=pkg_dir)
+
+
+if __name__ == "__main__":
+    import sys
+
+    generate_packages(sys.argv[1:])

scripts/pkg_gen/requirements.txt

@@ -0,0 +1,3 @@
+PyYaml >= 1.4.0
+click >= 8.1.3
+requests >= 2.31.0

scripts/pkg_gen/resources/auto-release.yaml

@@ -0,0 +1,36 @@
+name: Generate Release
+
+on:
+  repository_dispatch:
+    types: [create-release]
+
+jobs:
+  create_release:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v2
+
+      - name: Generate Release
+        run: |
+          # Extract necessary information from the event payload
+          REPO_OWNER=${{ github.event.client_payload.repo_owner }}
+          REPO_NAME=${{ github.event.client_payload.repo_name }}
+          RELEASE_TAG=${{ github.event.client_payload.release_tag }}
+          RELEASE_NAME=${{ github.event.client_payload.release_name }}
+          RELEASE_BODY=${{ github.event.client_payload.release_body }}
+
+          # Create a new release using the GitHub API
+          curl -X POST \
+            -H "Accept: application/vnd.github.v3+json" \
+            -H "Authorization: Bearer ${{ secrets.GITHUB_TOKEN }}" \
+            -d '{
+              "tag_name": "$RELEASE_TAG",
+              "target_commitish": "master",
+              "name": "$RELEASE_NAME",
+              "body": "$RELEASE_BODY",
+              "draft": false,
+              "prerelease": false
+            }' \
+            "https://api.github.com/repos/$REPO_OWNER/$REPO_NAME/releases"

scripts/pkg_gen/resources/nipype-auto-convert.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+import sys
+import os.path
+from warnings import warn
+from pathlib import Path
+from importlib import import_module
+import yaml
+import nipype
+import nipype2pydra
+from nipype2pydra.task import TaskConverter
+
+
+SPECS_DIR = Path(__file__).parent / "specs"
+PKG_ROOT = Path(__file__).parent.parent
+PKG_NAME = "CHANGEME"
+
+if ".dev" in nipype.__version__:
+    raise RuntimeError(
+        f"Cannot use a development version of Nipype {nipype.__version__}"
+    )
+
+if ".dev" in nipype2pydra.__version__:
+    warn(
+        f"using development version of nipype2pydra ({nipype2pydra.__version__}), "
+        f"development component will be dropped in {PKG_NAME} package version"
+    )
+n2p_version = nipype2pydra.__version__.split(".dev")[0]
+
+auto_version = f"{nipype.__version__}.{n2p_version}"
+
+
+# Insert specs dir into path so we can load callables modules
+sys.path.insert(0, str(SPECS_DIR))
+
+auto_init = f"# Auto-generated by {__file__}, do not edit as it will be overwritten\n\n"
+
+for fspath in SPECS_DIR.glob("**/*.yaml"):
+    with open(fspath) as f:
+        spec = yaml.load(f, Loader=yaml.SafeLoader)
+
+    rel_pkg_path = str(fspath.relative_to(SPECS_DIR)).replace(os.path.sep, ".")
+    callables = import_module(rel_pkg_path + "_callables")
+    module_name = fspath.name.lower()
+
+    converter = TaskConverter(
+        output_module=f"pydra.tasks.{PKG_NAME}.auto.{module_name}",
+        callables_module=callables,  # type: ignore
+        **spec
+    )
+    converter.generate(PKG_ROOT)
+    auto_init += f"from .{module_name} import {spec['task_name']}\n"
+
+
+with open(PKG_ROOT / "pydra" / "tasks" / PKG_NAME / "auto" / "_version.py", "w") as f:
+    f.write(f"""# Auto-generated by {__file__}, do not edit as it will be overwritten
+
+auto_version = {auto_version}
+""")
+
+with open(PKG_ROOT / "pydra" / "tasks" / PKG_NAME / "auto" / "__init__.py", "w") as f:
+    f.write(auto_init)

scripts/pkg_gen/resources/pkg_init.py

@@ -0,0 +1,22 @@
+"""
+This is a basic doctest demonstrating that the package and pydra can both be successfully
+imported.
+
+>>> import pydra.engine
+>>> import pydra.tasks.freesurfer
+"""
+try:
+    from ._version import __version__ as main_version
+except ImportError:
+    pass
+
+from .auto._version import auto_version  # Get version of 
+
+if ".dev" in main_version:
+    main_version, dev_version = main_version.split(".dev")
+else:
+    dev_version = None
+
+__version__ = main_version + "." + auto_version
+if dev_version:
+    __version__ += ".dev" + dev_version