Coverage for biobb_common/biobb_common/configuration/settings.py: 51%
145 statements
« prev ^ index » next coverage.py v7.4.3, created at 2024-03-13 17:26 +0000
1#!/usr/bin/env python3
3"""Settings loader module.
5This module contains the classes to read the different formats of the configuration files.
7The configuration files are composed by paths to the files and properties. There are several common properties for all
8the building blocks.
11Syntax:
12 - **property** (*dataType*) - (Default value) Short description.
14Available Workflow properties:
15 - **working_dir_path** (*str*) - (Current working dir) Workflow output directory.
16 - **can_write_console_log** (*bool*) - (True) Output log to console.
17 - **restart** (*bool*) - (False) Do not execute steps if output files are already created.
18 - **remove_tmp** (*bool*) - (True) Remove temporal files.
20Available common step properties: (Each Biobb step also has their specific properties)
21 - **can_write_console_log** (*bool*) - (True) Overwrite **can_write_console_log** workflow property on this step.
22 - **restart** (*bool*) - (False) Overwrite **restart** workflow property on this step.
23 - **remove_tmp** (*bool*) - (True) Overwrite **remove_tmp** workflow property on this step.
25Available common step properties for containerized applications:
26 - **container_path** (*string*) - (None) Path to the binary executable of your container.
27 - **container_image** (*string*) - (None) Container Image identifier.
28 - **container_volume_path** (*string*) - (None) Path to an internal directory in the container.
29 - **container_working_dir** (*string*) - (None) Path to the internal CWD in the container.
30 - **container_user_id** (*string*) - (None) User number id to be mapped inside the container.
31 - **container_shell_path** (*string*) - ("/bin/bash") Path to the binary executable of the container shell.
32"""
34import yaml
35import json
36import logging
37from pathlib import Path
38from biobb_common.tools import file_utils as fu
39from typing import Optional
41GALAXY_CHARACTER_MAP = {
42 "__gt__": ">",
43 "__lt__": "<",
44 "__sq__": "'",
45 "__dq__": '"',
46 "__ob__": "[",
47 "__cb__": "]",
48 "__oc__": "{",
49 "__cc__": "}",
50 "__cn__": "\n",
51 "__cr__": "\r",
52 "__tc__": "\t",
53 "__pd__": "#",
54}
57def trans_galaxy_charmap(input_str):
58 """Fixes escape characters introduced by Galaxy on Json inputs"""
59 for ch in GALAXY_CHARACTER_MAP:
60 input_str = input_str.replace(ch, GALAXY_CHARACTER_MAP[ch])
61 return input_str
64class ConfReader:
65 """Configuration file loader for yaml format files.
67 Args:
68 config (str): Path to the configuration [YAML|JSON] file or JSON string.
69 system (str): System name from the systems section in the configuration file.
70 """
72 def __init__(self, config: Optional[str] = None, system: Optional[str] = None):
73 if not config:
74 config = "{}"
75 self.config = config
76 self.system = system
77 self.properties = self._read_config()
78 if self.system:
79 self.properties[self.system]["working_dir_path"] = fu.get_working_dir_path(
80 self.properties[self.system].get("working_dir_path"),
81 restart=self.properties[self.system].get("restart", False),
82 )
83 else:
84 self.properties["working_dir_path"] = fu.get_working_dir_path(
85 self.properties.get("working_dir_path"),
86 restart=self.properties.get("restart", False),
87 )
89 def _read_config(self):
90 # Read just one step specified in the configuration file path
91 # i.e: Read just Editconf step from workflow_configuration.yaml file
92 # "/home/user/workflow_configuration.yaml#Editconf"
93 file_name_elements = str(self.config).split("#")
94 self.config = file_name_elements[0]
95 step = None
96 if len(file_name_elements) > 1:
97 step = file_name_elements[1]
98 try:
99 config_file = str(Path(self.config).resolve())
100 with open(config_file) as stream:
101 try:
102 config_dict = yaml.safe_load(stream)
103 except yaml.YAMLError:
104 config_dict = json.load(stream)
105 except Exception:
106 config_dict = json.loads(trans_galaxy_charmap(self.config))
107 if step:
108 return config_dict[step]
109 return config_dict
111 def get_working_dir_path(self) -> str:
112 if self.system:
113 return self.properties[self.system].get("working_dir_path")
115 return self.properties.get("working_dir_path")
117 def get_prop_dic(
118 self, prefix: Optional[str] = None, global_log: Optional[logging.Logger] = None
119 ) -> dict:
120 """get_prop_dic() returns the properties dictionary where keys are the
121 step names in the configuration YAML file and every value contains another
122 nested dictionary containing the keys and values of each step properties section.
123 All the paths in the system section are copied in each nested dictionary.
124 For each nested dictionary the following keys are added:
125 | **path** (*str*): Absolute path to the step working dir.
126 | **step** (*str*): Name of the step.
127 | **prefix** (*str*): Prefix if provided.
128 | **global_log** (*Logger object*): Log from the main workflow.
129 | **restart** (*bool*): Restart from previous execution.
130 | **remove_tmp** (*bool*): Remove temporal files.
132 Args:
133 prefix (str): Prefix if provided.
134 global_log (:obj:Logger): Log from the main workflow.
136 Returns:
137 dict: Dictionary of properties.
138 """
139 prop_dic = dict()
140 prefix = "" if prefix is None else prefix.strip()
142 # There is no step
143 if "paths" in self.properties or "properties" in self.properties:
144 prop_dic = dict()
145 if self.system:
146 prop_dic["path"] = str(
147 Path(self.properties[self.system]["working_dir_path"]).joinpath(
148 prefix
149 )
150 )
151 else:
152 prop_dic["path"] = str(
153 Path(self.properties["working_dir_path"]).joinpath(prefix)
154 )
155 prop_dic["step"] = None
156 prop_dic["prefix"] = prefix
157 prop_dic["global_log"] = global_log
158 prop_dic["system"] = self.system
159 if self.system:
160 prop_dic.update(self.properties[self.system].copy())
161 else:
162 prop_dic["working_dir_path"] = self.properties.get("working_dir_path")
163 prop_dic["restart"] = self.properties.get("restart", False)
164 prop_dic["remove_tmp"] = self.properties.get("remove_tmp", True)
166 if "properties" in self.properties and isinstance(
167 self.properties["properties"], dict
168 ):
169 prop_dic.update(self.properties["properties"].copy())
170 if self.system:
171 if self.properties[self.system].get("log_level", None):
172 prop_dic["log_level"] = self.properties[self.system][
173 "log_level"
174 ]
175 else:
176 if self.properties.get("log_level", None):
177 prop_dic["log_level"] = self.properties["log_level"]
178 # There is step name
179 else:
180 for key in self.properties:
181 if isinstance(self.properties[key], dict):
182 if (
183 "paths" in self.properties[key]
184 or "properties" in self.properties[key]
185 ):
186 prop_dic[key] = dict()
187 if self.system:
188 prop_dic[key]["path"] = str(
189 Path(
190 self.properties[self.system]["working_dir_path"]
191 ).joinpath(prefix, key)
192 )
193 else:
194 prop_dic[key]["path"] = str(
195 Path(self.properties["working_dir_path"]).joinpath(
196 prefix, key
197 )
198 )
199 prop_dic[key]["step"] = key
200 prop_dic[key]["prefix"] = prefix
201 prop_dic[key]["global_log"] = global_log
202 prop_dic[key]["system"] = self.system
203 if self.system:
204 prop_dic[key].update(self.properties[self.system].copy())
205 else:
206 prop_dic[key]["working_dir_path"] = self.properties.get(
207 "working_dir_path"
208 )
209 prop_dic[key][
210 "can_write_console_log"
211 ] = self.properties.get("can_write_console_log", True)
212 prop_dic[key]["restart"] = self.properties.get(
213 "restart", False
214 )
215 prop_dic[key]["remove_tmp"] = self.properties.get(
216 "remove_tmp", True
217 )
219 if ("properties" in self.properties[key]) and isinstance(
220 self.properties[key]["properties"], dict
221 ):
222 if self.system:
223 if self.properties[self.system].get("log_level", None):
224 prop_dic[key]["log_level"] = self.properties[
225 self.system
226 ]["log_level"]
227 prop_dic[key]["can_write_console_log"] = self.properties[
228 self.system
229 ].get("can_write_console_log", True)
230 else:
231 if self.properties.get("log_level", None):
232 prop_dic[key]["log_level"] = self.properties[
233 "log_level"
234 ]
235 prop_dic[key][
236 "can_write_console_log"
237 ] = self.properties.get("can_write_console_log", True)
238 prop_dic[key].update(self.properties[key]["properties"].copy())
240 # There is no step name and there is no properties or paths key return input
241 if not prop_dic:
242 prop_dic = dict()
243 prop_dic.update(self.properties)
244 if self.system:
245 prop_dic["path"] = str(
246 Path(self.properties[self.system]["working_dir_path"]).joinpath(
247 prefix
248 )
249 )
250 else:
251 prop_dic["path"] = str(
252 Path(self.properties["working_dir_path"]).joinpath(prefix)
253 )
254 prop_dic["step"] = None
255 prop_dic["prefix"] = prefix
256 prop_dic["global_log"] = global_log
257 prop_dic["system"] = self.system
258 if self.system:
259 prop_dic.update(self.properties[self.system].copy())
260 else:
261 prop_dic["working_dir_path"] = self.properties.get("working_dir_path")
262 prop_dic["can_write_console_log"] = self.properties.get(
263 "can_write_console_log", True
264 )
265 prop_dic["restart"] = self.properties.get("restart", False)
266 prop_dic["remove_tmp"] = self.properties.get("remove_tmp", True)
268 return prop_dic
270 def get_paths_dic(self, prefix: Optional[str] = None) -> dict:
271 """get_paths_dic() returns the paths dictionary where keys are the
272 step names in the configuration YAML file and every value contains another
273 nested dictionary containing the keys and values of each step paths section.
274 All the paths starting with 'dependency' are resolved. If the path starts
275 with the string 'file:' nothing is done, however if the path starts with
276 any other string path is prefixed with the absolute step path.
278 Args:
279 prefix (str): Prefix if provided.
281 Returns:
282 dict: Dictionary of paths.
283 """
284 prop_dic = dict()
285 prefix = "" if prefix is None else prefix.strip()
286 # Filtering just paths
287 # Properties without step name
288 if "paths" in self.properties:
289 step = False
290 prop_dic = self.properties["paths"].copy()
292 # Properties with name
293 else:
294 step = True
296 for key in self.properties:
297 if isinstance(self.properties[key], dict):
298 if "paths" in self.properties[key]:
299 prop_dic[key] = self.properties[key]["paths"].copy()
300 else:
301 prop_dic[key] = {}
303 # Solving dependencies and adding workflow and step path
304 # Properties without step name: Do not solving dependencies
305 if not step:
306 for key2, value in prop_dic.items():
307 if isinstance(value, str) and value.startswith("file:"):
308 prop_dic[key2] = value.split(":")[1]
309 else:
310 if self.system:
311 prop_dic[key2] = str(
312 Path(
313 self.properties[self.system]["working_dir_path"]
314 ).joinpath(prefix, key, value)
315 )
316 else:
317 prop_dic[key2] = str(
318 Path(self.properties["working_dir_path"]).joinpath(
319 prefix, value
320 )
321 )
323 # Properties with step name
324 else:
325 for key in prop_dic:
326 for key2, value in prop_dic[key].items():
327 if isinstance(value, str) and value.startswith("dependency"):
328 while isinstance(value, str) and value.startswith("dependency"):
329 dependency_step = value.split("/")[1]
330 value = prop_dic[value.split("/")[1]][value.split("/")[2]]
331 if self.properties.get(self.system):
332 prop_dic[key][key2] = str(
333 Path(
334 self.properties[self.system]["working_dir_path"]
335 ).joinpath(prefix, dependency_step, value)
336 )
337 else:
338 prop_dic[key][key2] = str(
339 Path(self.properties["working_dir_path"]).joinpath(
340 prefix, dependency_step, value
341 )
342 )
343 elif isinstance(value, str) and value.startswith("file:"):
344 prop_dic[key][key2] = value.split(":")[1]
345 else:
346 if self.system:
347 prop_dic[key][key2] = str(
348 Path(
349 self.properties[self.system]["working_dir_path"]
350 ).joinpath(prefix, key, value)
351 )
352 else:
353 prop_dic[key][key2] = str(
354 Path(self.properties["working_dir_path"]).joinpath(
355 prefix, key, value
356 )
357 )
359 return prop_dic