diff env/lib/python3.9/site-packages/gxformat2/model.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/model.py	Mon Mar 22 18:12:50 2021 +0000
@@ -0,0 +1,181 @@
+"""Abstractions for dealing with Format2 data."""
+from typing import cast, Dict, List, Union
+
+DictOrList = Union[Dict, List]
+
+
+def convert_dict_to_id_list_if_needed(
+    dict_or_list: DictOrList,
+    add_label: bool = False,
+    mutate: bool = False,
+) -> list:
+    """Convert a list or dict to a list with keys embedded.
+
+    If `add_label` is True, embed dict keys as 'label' attribute
+    else 'id'.
+    """
+    if isinstance(dict_or_list, dict):
+        rval = []
+        for key, value in dict_or_list.items():
+            if not isinstance(value, dict):
+                value = {"type": value}
+            if not mutate:
+                value = value.copy()
+            if add_label:
+                if value.get("label") is None:
+                    value["label"] = key
+            else:
+                value["id"] = key
+            rval.append(value)
+    else:
+        rval = cast(list, dict_or_list)
+    return rval
+
+
+def with_step_ids(steps: list, inputs_offset: int = 0):
+    """Walk over a list of steps and ensure the steps have a numeric id if otherwise missing."""
+    assert isinstance(steps, list)
+    new_steps = []
+    for i, step in enumerate(steps):
+        if "id" not in step:
+            step = step.copy()
+            step["id"] = i + inputs_offset
+        assert step["id"] is not None
+        new_steps.append(step)
+    return new_steps
+
+
+def ensure_step_position(step: dict, order_index: int):
+    """Ensure step contains a position definition.
+
+    Modifies the input step dictionary.
+    """
+    if "position" not in step:
+        step["position"] = {
+            "left": 10 * order_index,
+            "top": 10 * order_index
+        }
+
+
+def native_input_to_format2_type(step: dict, tool_state: dict) -> str:
+    """Return a Format2 input type ('type') from a native input step dictionary."""
+    module_type = step.get("type")
+    if module_type == 'data_collection_input':
+        format2_type = 'collection'
+    elif module_type == 'data_input':
+        format2_type = 'data'
+    elif module_type == "parameter_input":
+        native_type = cast(str, tool_state.get("parameter_type"))
+        format2_type = native_type
+        if native_type == "integer":
+            format2_type = "int"
+        elif native_type == "text":
+            format2_type = "string"
+    return format2_type
+
+
+def inputs_as_normalized_steps(workflow_dict):
+    """Return workflow inputs to a steps in array.
+
+    Normalize Format2 inputs. `workflow_dict` is a Format 2 representation of
+    a workflow. This method does not modify `workflow_dict`.
+    """
+    if "inputs" not in workflow_dict:
+        return []
+
+    inputs = workflow_dict.get("inputs", [])
+    new_steps = []
+    inputs = convert_dict_to_id_list_if_needed(inputs)
+    for input_def_raw in with_step_ids(inputs):
+        input_def = input_def_raw.copy()
+
+        if "label" in input_def and "id" in input_def:
+            raise Exception("label and id are aliases for inputs, may only define one")
+        if "label" not in input_def and "id" not in input_def:
+            raise Exception("Input must define a label.")
+
+        raw_label = input_def.pop("label", None)
+        raw_id = input_def.pop("id", None)
+        label = raw_label or raw_id
+
+        if label is None:
+            raise Exception("Input label must not be empty.")
+
+        step_type = input_def.pop("type", "data")
+        if step_type == "File":
+            step_type = "data"
+        elif step_type == "integer":
+            step_type = "int"
+        elif step_type == "text":
+            step_type = "string"
+
+        step_def = input_def
+        step_def.update({
+            "type": step_type,
+            "id": label,
+        })
+        new_steps.append(step_def)
+
+    return new_steps
+
+
+def inputs_as_native_steps(workflow_dict: dict):
+    """Return workflow inputs to a steps in array - like in native Galaxy.
+
+    Convert Format2 types into native ones. `workflow_dict` is a Format 2
+    representation of a workflow. This method does not modify `workflow_dict`.
+    """
+    if "inputs" not in workflow_dict:
+        return []
+
+    inputs = workflow_dict.get("inputs", [])
+    new_steps = []
+    inputs = convert_dict_to_id_list_if_needed(inputs)
+    for input_def_raw in inputs:
+        input_def = input_def_raw.copy()
+
+        if "label" in input_def and "id" in input_def:
+            raise Exception("label and id are aliases for inputs, may only define one")
+        if "label" not in input_def and "id" not in input_def:
+            raise Exception("Input must define a label.")
+
+        raw_label = input_def.pop("label", None)
+        raw_id = input_def.pop("id", None)
+        label = raw_label or raw_id
+
+        if label is None:
+            raise Exception("Input label must not be empty.")
+
+        input_type = input_def.pop("type", "data")
+        if input_type in ["File", "data", "data_input"]:
+            step_type = "data_input"
+        elif input_type in ["collection", "data_collection", "data_collection_input"]:
+            step_type = "data_collection_input"
+        elif input_type in ["text", "string", "integer", "int", "float", "color", "boolean"]:
+            step_type = "parameter_input"
+            format2_type = input_type
+            if format2_type == "int":
+                native_type = "integer"
+            elif format2_type == "string":
+                native_type = "text"
+            else:
+                native_type = format2_type
+            input_def["parameter_type"] = native_type
+        else:
+            raise Exception("Unknown input type [%s] encountered." % input_type)
+
+        step_def = input_def
+        step_def.update({
+            "type": step_type,
+            "label": label,
+        })
+        new_steps.append(step_def)
+
+    return new_steps
+
+
+def outputs_as_list(as_python: dict) -> list:
+    """Extract outputs from Format2 rep as list."""
+    outputs = as_python.get("outputs", [])
+    outputs = convert_dict_to_id_list_if_needed(outputs)
+    return outputs