Mercurial > repos > shellac > sam_consensus_v3

comparison env/lib/python3.9/site-packages/gxformat2/model.py @ 0:4f3585e2f14b draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author shellac
date Mon, 22 Mar 2021 18:12:50 +0000
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:4f3585e2f14b
1 """Abstractions for dealing with Format2 data."""
2 from typing import cast, Dict, List, Union
3
4 DictOrList = Union[Dict, List]
5
6
7 def convert_dict_to_id_list_if_needed(
8 dict_or_list: DictOrList,
9 add_label: bool = False,
10 mutate: bool = False,
11 ) -> list:
12 """Convert a list or dict to a list with keys embedded.
13
14 If `add_label` is True, embed dict keys as 'label' attribute
15 else 'id'.
16 """
17 if isinstance(dict_or_list, dict):
18 rval = []
19 for key, value in dict_or_list.items():
20 if not isinstance(value, dict):
21 value = {"type": value}
22 if not mutate:
23 value = value.copy()
24 if add_label:
25 if value.get("label") is None:
26 value["label"] = key
27 else:
28 value["id"] = key
29 rval.append(value)
30 else:
31 rval = cast(list, dict_or_list)
32 return rval
33
34
35 def with_step_ids(steps: list, inputs_offset: int = 0):
36 """Walk over a list of steps and ensure the steps have a numeric id if otherwise missing."""
37 assert isinstance(steps, list)
38 new_steps = []
39 for i, step in enumerate(steps):
40 if "id" not in step:
41 step = step.copy()
42 step["id"] = i + inputs_offset
43 assert step["id"] is not None
44 new_steps.append(step)
45 return new_steps
46
47
48 def ensure_step_position(step: dict, order_index: int):
49 """Ensure step contains a position definition.
50
51 Modifies the input step dictionary.
52 """
53 if "position" not in step:
54 step["position"] = {
55 "left": 10 * order_index,
56 "top": 10 * order_index
57 }
58
59
60 def native_input_to_format2_type(step: dict, tool_state: dict) -> str:
61 """Return a Format2 input type ('type') from a native input step dictionary."""
62 module_type = step.get("type")
63 if module_type == 'data_collection_input':
64 format2_type = 'collection'
65 elif module_type == 'data_input':
66 format2_type = 'data'
67 elif module_type == "parameter_input":
68 native_type = cast(str, tool_state.get("parameter_type"))
69 format2_type = native_type
70 if native_type == "integer":
71 format2_type = "int"
72 elif native_type == "text":
73 format2_type = "string"
74 return format2_type
75
76
77 def inputs_as_normalized_steps(workflow_dict):
78 """Return workflow inputs to a steps in array.
79
80 Normalize Format2 inputs. `workflow_dict` is a Format 2 representation of
81 a workflow. This method does not modify `workflow_dict`.
82 """
83 if "inputs" not in workflow_dict:
84 return []
85
86 inputs = workflow_dict.get("inputs", [])
87 new_steps = []
88 inputs = convert_dict_to_id_list_if_needed(inputs)
89 for input_def_raw in with_step_ids(inputs):
90 input_def = input_def_raw.copy()
91
92 if "label" in input_def and "id" in input_def:
93 raise Exception("label and id are aliases for inputs, may only define one")
94 if "label" not in input_def and "id" not in input_def:
95 raise Exception("Input must define a label.")
96
97 raw_label = input_def.pop("label", None)
98 raw_id = input_def.pop("id", None)
99 label = raw_label or raw_id
100
101 if label is None:
102 raise Exception("Input label must not be empty.")
103
104 step_type = input_def.pop("type", "data")
105 if step_type == "File":
106 step_type = "data"
107 elif step_type == "integer":
108 step_type = "int"
109 elif step_type == "text":
110 step_type = "string"
111
112 step_def = input_def
113 step_def.update({
114 "type": step_type,
115 "id": label,
116 })
117 new_steps.append(step_def)
118
119 return new_steps
120
121
122 def inputs_as_native_steps(workflow_dict: dict):
123 """Return workflow inputs to a steps in array - like in native Galaxy.
124
125 Convert Format2 types into native ones. `workflow_dict` is a Format 2
126 representation of a workflow. This method does not modify `workflow_dict`.
127 """
128 if "inputs" not in workflow_dict:
129 return []
130
131 inputs = workflow_dict.get("inputs", [])
132 new_steps = []
133 inputs = convert_dict_to_id_list_if_needed(inputs)
134 for input_def_raw in inputs:
135 input_def = input_def_raw.copy()
136
137 if "label" in input_def and "id" in input_def:
138 raise Exception("label and id are aliases for inputs, may only define one")
139 if "label" not in input_def and "id" not in input_def:
140 raise Exception("Input must define a label.")
141
142 raw_label = input_def.pop("label", None)
143 raw_id = input_def.pop("id", None)
144 label = raw_label or raw_id
145
146 if label is None:
147 raise Exception("Input label must not be empty.")
148
149 input_type = input_def.pop("type", "data")
150 if input_type in ["File", "data", "data_input"]:
151 step_type = "data_input"
152 elif input_type in ["collection", "data_collection", "data_collection_input"]:
153 step_type = "data_collection_input"
154 elif input_type in ["text", "string", "integer", "int", "float", "color", "boolean"]:
155 step_type = "parameter_input"
156 format2_type = input_type
157 if format2_type == "int":
158 native_type = "integer"
159 elif format2_type == "string":
160 native_type = "text"
161 else:
162 native_type = format2_type
163 input_def["parameter_type"] = native_type
164 else:
165 raise Exception("Unknown input type [%s] encountered." % input_type)
166
167 step_def = input_def
168 step_def.update({
169 "type": step_type,
170 "label": label,
171 })
172 new_steps.append(step_def)
173
174 return new_steps
175
176
177 def outputs_as_list(as_python: dict) -> list:
178 """Extract outputs from Format2 rep as list."""
179 outputs = as_python.get("outputs", [])
180 outputs = convert_dict_to_id_list_if_needed(outputs)
181 return outputs