Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/gxformat2/main.py @ 2:6af9afd405e9 draft
"planemo upload commit 0a63dd5f4d38a1f6944587f52a8cd79874177fc1"
author | shellac |
---|---|
date | Thu, 14 May 2020 14:56:58 -0400 |
parents | 26e78fe6e8c4 |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.7/site-packages/gxformat2/main.py Thu May 14 14:56:58 2020 -0400 @@ -0,0 +1,61 @@ +"""Module containing :func:`convert_and_import_workflow`.""" +import os + +from ._yaml import ordered_load +from .converter import python_to_workflow, yaml_to_workflow +from .interface import BioBlendImporterGalaxyInterface + + +def convert_and_import_workflow(has_workflow, **kwds): + """Conversion and import Format 2 workflows into a target Galaxy instance.""" + galaxy_interface = kwds.get("galaxy_interface", None) + if galaxy_interface is None: + galaxy_interface = BioBlendImporterGalaxyInterface(**kwds) + + source_type = kwds.get("source_type", None) + workflow_directory = kwds.get("workflow_directory", None) + if source_type == "path": + workflow_path = has_workflow + if workflow_directory is None: + workflow_directory = os.path.dirname(has_workflow) + with open(workflow_path, "r") as f: + has_workflow = ordered_load(f) + + if workflow_directory is not None: + workflow_directory = os.path.abspath(workflow_directory) + + convert = kwds.get("convert", True) + raw_yaml = kwds.get("raw_yaml", False) + if raw_yaml and convert: + raise Exception("Incompatible options selected.") + if convert: + if isinstance(has_workflow, dict): + workflow = python_to_workflow(has_workflow, galaxy_interface, workflow_directory) + else: + workflow = yaml_to_workflow(has_workflow, galaxy_interface, workflow_directory) + else: + workflow = has_workflow + if not isinstance(workflow, dict) and not raw_yaml: + workflow = ordered_load(workflow) + else: + workflow = {"yaml_content": workflow} + + name = kwds.get("name", None) + if name is not None: + workflow["name"] = name + publish = kwds.get("publish", False) + exact_tools = kwds.get("exact_tools", False) + fill_defaults = kwds.get("fill_defaults", True) + import_kwds = { + "fill_defaults": fill_defaults + } + if publish: + import_kwds["publish"] = True + if exact_tools: + import_kwds["exact_tools"] = True + return galaxy_interface.import_workflow(workflow, **import_kwds) + + +__all__ = ( + 'convert_and_import_workflow', +)