Mercurial > repos > shellac > sam_consensus_v3
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