Add --from_workflow to shed_init.

Builds out a repository_dependencies.xml from the specified workflow. If needed, this also copies the workflow into the target repository.

Closes #118.

planemo/commands/cmd_shed_init.py

@@ -9,13 +9,12 @@
 
 @click.command("shed_init")
 @options.optional_project_arg()
-# TODO:
-# @click.option(
-#    "--from_workflow",
-#    type=click.Path(exists=True, file_okay=True, resolve_path=True),
-#    help=('Attempt to generate repository dependencies from specified '
-#          'workflow.')
-# )
+@click.option(
+    "--from_workflow",
+    type=click.Path(exists=True, file_okay=True, resolve_path=True),
+    help=('Attempt to generate repository dependencies from specified '
+          'workflow.')
+)
 @click.option(
     "--description",
     help='Specify repository description for .shed.yml.'

planemo/shed.py

@@ -1,6 +1,7 @@
 import fnmatch
 import glob
 import hashlib
+import json
 import os
 import tarfile
 from tempfile import (
@@ -106,6 +107,26 @@ def shed_init(ctx, path, **kwds):
         return 1
 
     _create_shed_config(ctx, shed_config_path, **kwds)
+    repo_dependencies_path = os.path.join(path, "repository_dependencies.xml")
+    from_workflow = kwds.get("from_workflow", None)
+
+    if from_workflow:
+        workflow_name = os.path.basename(from_workflow)
+        workflow_target = os.path.join(path, workflow_name)
+        if not os.path.exists(workflow_target):
+            shutil.copyfile(from_workflow, workflow_target)
+
+        if not can_write_to_path(repo_dependencies_path, **kwds):
+            return 1
+
+        repo_pairs = _parse_repos_from_workflow(from_workflow)
+        contents = '<repositories description="">'
+        line_template = '  <repository owner="%s" name="%s" />'
+        for (owner, name) in repo_pairs:
+            contents += line_template % (owner, name)
+        contents += "</repositories>"
+        open(repo_dependencies_path, "w").write(contents)
+
     return 0
 
 
@@ -369,6 +390,29 @@ def _create_shed_config(ctx, path, **kwds):
         yaml.dump(config, f)
 
 
+def _parse_repos_from_workflow(path):
+    workflow_json = json.load(open(path, "r"))
+    steps = workflow_json["steps"]
+    tool_ids = set()
+    for value in steps.values():
+        step_type = value["type"]
+        if step_type != "tool":
+            continue
+        tool_id = value["tool_id"]
+        if "/repos/" in tool_id:
+            tool_ids.add(tool_id)
+
+    repo_pairs = set()
+    for tool_id in tool_ids:
+        tool_repo_info = tool_id.split("/repos/", 1)[1]
+        tool_repo_parts = tool_repo_info.split("/")
+        owner = tool_repo_parts[0]
+        name = tool_repo_parts[1]
+        repo_pairs.add((owner, name))
+
+    return repo_pairs
+
+
 def _find_raw_repositories(path, **kwds):
     name = kwds.get("name", None)
     recursive = kwds.get("recursive", False)

tests/test_shed_init.py

@@ -2,7 +2,16 @@
 
 import yaml
 
-from .test_utils import CliShedTestCase
+from .test_utils import (
+    CliShedTestCase,
+    TEST_REPOS_DIR,
+)
+
+TEST_WORKFLOW_PATH = os.path.join(
+    TEST_REPOS_DIR,
+    "workflow_1",
+    "blast_top_hit_species.ga"
+)
 
 
 class ShedInitTestCase(CliShedTestCase):
@@ -50,3 +59,19 @@ def test_more_options(self):
             assert len(categories) == 3
             assert "SAM" in categories
             assert "Statistics" in categories
+
+    def test_from_workflow(self):
+        with self._isolate() as f:
+            init_command = ["shed_init", "--owner", "iuc"]
+            init_command += ["--category", "Sequence Analysis"]
+            init_command += ["--name", "blasthits"]
+            init_command += ["--from_workflow", TEST_WORKFLOW_PATH]
+            self._check_exit_code(init_command)
+
+            repo_deps_path = os.path.join(f, "repository_dependencies.xml")
+            wf_path = os.path.join(f, "blast_top_hit_species.ga")
+            assert os.path.exists(repo_deps_path)
+            assert os.path.exists(wf_path)
+
+            # lint repository as a way of verifying repository_dependencies
+            self._check_exit_code(["shed_lint"])