Mercurial > repos > shellac > sam_consensus_v3
diff env/lib/python3.9/site-packages/gxformat2/main.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.9/site-packages/gxformat2/main.py Mon Mar 22 18:12:50 2021 +0000 @@ -0,0 +1,61 @@ +"""Module containing :func:`convert_and_import_workflow`.""" +import os + +from .converter import python_to_workflow, yaml_to_workflow +from .interface import BioBlendImporterGalaxyInterface +from .yaml import ordered_load + + +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', +)