Coverage for python/lsst/ctrl/bps/bps_config.py: 97%
211 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-30 09:07 +0000
1# This file is part of ctrl_bps.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <https://www.gnu.org/licenses/>.
28"""Configuration class that adds order to searching sections for value,
29expands environment variables and other config variables.
30"""
32__all__ = ["BPS_DEFAULTS", "BPS_SEARCH_ORDER", "BpsConfig", "BpsFormatter"]
35import copy
36import logging
37import os
38import re
39import string
40from os.path import expandvars, normpath
41from typing import Any
43from lsst.daf.butler import Config
44from lsst.resources import ResourcePath
45from lsst.utils import doImport
47from .bps_utils import bps_eval
49_LOG = logging.getLogger(__name__)
51# Using lsst.daf.butler.Config to resolve possible includes.
52BPS_DEFAULTS = Config(ResourcePath("resource://lsst.ctrl.bps/etc/bps_defaults.yaml")).toDict()
54BPS_SEARCH_ORDER = ["bps_cmdline", "payload", "cluster", "pipetask", "site", "cloud", "bps_defined"]
56# Need a string that won't be a valid default value
57# to indicate whether default was defined for search.
58# And None is a valid default value.
59_NO_SEARCH_DEFAULT_VALUE = "__NO_SEARCH_DEFAULT_VALUE__"
62class BpsFormatter(string.Formatter):
63 """String formatter class that allows BPS config search options."""
65 def get_field(self, field_name, args, kwargs):
66 _, val = args[0].search(field_name, opt=args[1])
67 return val, field_name
69 def get_value(self, key, args, kwargs):
70 _, val = args[0].search(key, opt=args[1])
71 return val
74class BpsConfig(Config):
75 """Contains the configuration for a BPS submission.
77 Parameters
78 ----------
79 other : `str`, `dict`, `~lsst.daf.butler.Config`, `BpsConfig`
80 Path to a YAML file or a dict/Config/BpsConfig containing configuration
81 to copy.
82 search_order : `list` [`str`], optional
83 Root section names in the order in which they should be searched.
84 defaults : `str`, `dict`, `~lsst.daf.butler.Config`, optional
85 Default settings that will be used to prepopulate the config.
86 If the WMS service default settings are available, they will be added
87 afterwards. WMS settings takes precedence over provided defaults.
88 wms_service_class_fqn : `str`, optional
89 Fully qualified name of the WMS service class to use to get plugin's
90 specific default settings. If `None` (default), the WMS service
91 class provided by
93 1. ``other`` config,
94 2. environmental variable ``BPS_WMS_SERVICE_CLASS``,
95 3. default settings
97 will be used instead. The list above also reflects the priorities
98 if the WMS service class is defined in multiple places. For example,
99 the name of service class found in ``other`` takes precedence over
100 the name of the service class provided by the BPS_SERVICE_CLASS and/or
101 the default settings.
103 Raises
104 ------
105 ValueError
106 Raised if the class cannot be instantiated from the provided object.
107 """
109 def __init__(self, other, search_order=None, defaults=None, wms_service_class_fqn=None):
110 # In BPS config, the same setting can be defined multiple times in
111 # different sections. The sections are search in a pre-defined
112 # order. Hence, a value which is found first effectively overrides
113 # values in later sections, if any. To achieve this goal,
114 # the special methods __getitem__ and __contains__ were redefined to
115 # use a custom search function internally. For this reason we can't
116 # use super().__init__(other) as the super class defines its own
117 # __getitem__ which is utilized during the initialization process (
118 # e.g. in expressions like self[<key>]). However, this function will
119 # be overridden by the one defined here, in the subclass. Instead
120 # we just initialize internal data structures and populate them
121 # using the inherited update() method which does not rely on super
122 # class __getitem__ method.
123 super().__init__()
125 try:
126 other_config = Config(other)
127 except Exception as exc:
128 raise ValueError(f"A BpsConfig could not be loaded from other: {other}") from exc
130 config = Config()
132 # Pre-populate the config with default settings if any were provided
133 # by the caller. Include WMS plugin specific defaults and/or
134 # overrides as well if available.
135 if defaults:
136 config.update(defaults)
138 # If the WMS service class was not specified explicitly by the
139 # caller, try to use the value provided by either:
140 #
141 # 1. 'other' config,
142 # 2. environmental variable BPS_WMS_SERVICE_CLASS,
143 # 3. default settings
144 #
145 # (in decreasing priority).
146 if wms_service_class_fqn is None:
147 wms_service_class_fqn = other_config.get(
148 "wmsServiceClass",
149 os.environ.get("BPS_WMS_SERVICE_CLASS", config.get("wmsServiceClass")),
150 )
151 try:
152 wms_service_class = doImport(wms_service_class_fqn)
153 except TypeError:
154 # Do not die if the WMS service class is still not set.
155 pass
156 else:
157 wms_service = wms_service_class({})
158 wms_defaults = wms_service.defaults
159 if wms_defaults:
160 config.update(wms_defaults)
162 # Set the service class to the one which was defaults was used.
163 config["wmsServiceClass"] = wms_service_class_fqn
165 # Include values and/or apply overrides from 'other' config.
166 config.update(other_config)
167 self.update(config)
169 if isinstance(other, BpsConfig):
170 self.formatter = copy.deepcopy(other.formatter)
171 self.search_order = copy.deepcopy(other.search_order) if search_order is None else search_order
172 else:
173 self.formatter = BpsFormatter()
174 self.search_order = BPS_SEARCH_ORDER if search_order is None else search_order
176 # Make sure search sections exist.
177 for key in self.search_order:
178 if not Config.__contains__(self, key):
179 self[key] = {}
181 def copy(self):
182 """Make a copy of config.
184 Returns
185 -------
186 copy : `lsst.ctrl.bps.BpsConfig`
187 A duplicate of itself.
188 """
189 return BpsConfig(self)
191 def get(self, key, default=""):
192 """Return the value for key if key is in the config, else default.
194 If default is not given, it defaults to an empty string.
196 Parameters
197 ----------
198 key : `str`
199 Key to look for in config.
200 default : `~typing.Any`, optional
201 Default value to return if the key is not in the config.
203 Returns
204 -------
205 val : Any
206 Value from config if found, default otherwise.
208 Notes
209 -----
210 The provided default value (an empty string) was chosen to maintain
211 the internal consistency with other methods of the class.
212 """
213 _, val = self.search(key, opt={"default": default})
214 return val
216 def __getitem__(self, name):
217 """Return the value from the config for the given name.
219 Parameters
220 ----------
221 name : `str`
222 Key to look for in config
224 Returns
225 -------
226 val : `str`, `int`, `lsst.ctrl.bps.BpsConfig`, ...
227 Value from config if found.
228 """
229 _, val = self.search(name, {})
231 return val
233 def __contains__(self, name):
234 """Check whether name is in config.
236 Parameters
237 ----------
238 name : `str`
239 Key to look for in config.
241 Returns
242 -------
243 found : `bool`
244 Whether name was in config or not.
245 """
246 found, _ = self.search(name, {})
247 return found
249 def search(self, key, opt=None):
250 """Search for key using given opt following hierarchy rules.
252 Search hierarchy rules: current values, a given search object, and
253 search order of config sections.
255 Parameters
256 ----------
257 key : `str`
258 Key to look for in config.
259 opt : `dict` [`str`, `~typing.Any`], optional
260 Options dictionary to use while searching. All are optional.
262 ``"curvals"``
263 Means to pass in values for search order key
264 (curr_<sectname>) or variable replacements.
265 (`dict`, optional)
266 ``"default"``
267 Value to return if not found. (`~typing.Any`, optional)
268 ``"replaceEnvVars"``
269 Kept for backward compatibility.
270 See ``replaceEnvShell2Bps``.
271 ``"replaceEnvShell2Bps"``
272 If search result is string, whether to replace environment
273 variables inside it with special placeholder (<ENV:name>).
274 By default set to False. (`bool`)
275 ``"replaceEnvBps2Shell"``
276 If search result is string, whether to replace environment
277 placeholder inside it with shell environment syntax.
278 By default set to False. (`bool`)
279 ``"expandEnvVars"``
280 If search result is string, whether to replace environment
281 placeholder inside it with current environment value.
282 By default set to False. (`bool`)
283 ``"replaceVars"``
284 If search result is string, whether to replace variables
285 inside it. By default set to True. (`bool`)
286 ``"required"``
287 If replacing variables, whether to raise exception if
288 variable is undefined. By default set to False. (`bool`)
290 Returns
291 -------
292 found : `bool`
293 Whether name was in config or not.
294 value : `str`, `int`, `lsst.ctrl.bps.BpsConfig`, ...
295 Value from config if found.
296 """
297 _LOG.debug("search: initial key = '%s', opt = '%s'", key, opt)
299 if opt is None:
300 opt = {}
302 found = False
303 value = ""
305 # start with stored current values
306 curvals = None
307 if Config.__contains__(self, "current"): 307 ↛ 308line 307 didn't jump to line 308 because the condition on line 307 was never true
308 curvals = copy.deepcopy(Config.__getitem__(self, "current"))
309 else:
310 curvals = {}
312 # override with current values passed into function if given
313 if "curvals" in opt:
314 for ckey, cval in list(opt["curvals"].items()):
315 _LOG.debug("using specified curval %s = %s", ckey, cval)
316 curvals[ckey] = cval
318 _LOG.debug("curvals = %s", curvals)
320 # There's a problem with the searchobj being a BpsConfig
321 # and its handling of __getitem__. Until that part of
322 # BpsConfig is rewritten, force the searchobj to a Config.
323 if "searchobj" in opt:
324 opt["searchobj"] = Config(opt["searchobj"])
326 if key in curvals:
327 _LOG.debug("found %s in curvals", key)
328 found = True
329 value = curvals[key]
330 elif "searchobj" in opt and key in opt["searchobj"]:
331 found = True
332 value = opt["searchobj"][key]
333 else:
334 for sect in self.search_order:
335 if Config.__contains__(self, sect): 335 ↛ 350line 335 didn't jump to line 350 because the condition on line 335 was always true
336 _LOG.debug("Searching '%s' section for key '%s'", sect, key)
337 search_sect = Config.__getitem__(self, sect)
338 if "curr_" + sect in curvals:
339 currkey = curvals["curr_" + sect]
340 _LOG.debug("currkey for section %s = %s", sect, currkey)
341 if Config.__contains__(search_sect, currkey):
342 search_sect = Config.__getitem__(search_sect, currkey)
344 _LOG.debug("%s %s", key, search_sect)
345 if Config.__contains__(search_sect, key):
346 found = True
347 value = Config.__getitem__(search_sect, key)
348 break
349 else:
350 _LOG.debug("Missing search section '%s' while searching for '%s'", sect, key)
352 # lastly check root values
353 if not found:
354 _LOG.debug("Searching root section for key '%s'", key)
355 if Config.__contains__(self, key):
356 found = True
357 value = Config.__getitem__(self, key)
358 _LOG.debug("root value='%s'", value)
359 else:
360 _LOG.debug("not in root section for key='%s'", key)
361 _LOG.debug(Config(self))
363 if not found and "default" in opt:
364 value = opt["default"]
365 found = True # ????
367 if not found and opt.get("required", False):
368 print(f"\n\nError: search for {key} failed")
369 print("\tcurrent = ", self.get("current"))
370 print("\topt = ", opt)
371 print("\tcurvals = ", curvals)
372 print("\n\n")
373 raise KeyError(f"Error: Search failed {key}")
375 _LOG.debug("found=%s, value=%s", found, value)
377 _LOG.debug("opt=%s %s", opt, type(opt))
378 if found and isinstance(value, str):
379 value = self.modify_value(key, value, opt)
380 _LOG.debug("after modify_value=%s", value)
382 if found and isinstance(value, Config):
383 value = BpsConfig(value, search_order=[])
385 return found, value
387 def replace_vars(self, value: str, opt: dict[str, Any]) -> str:
388 """Replace variables in string with values except those
389 in opt['skipNames'].
391 Parameters
392 ----------
393 value : `str`
394 Value in which to replace variables.
395 opt : `dict` [`str`, Any]
396 Options to be used when searching and replacing values.
397 In particular "skipNames" lists variable names to
398 not replace.
399 """
400 # default only applies to original search key
401 # Instead of doing deep copies of opt (especially with
402 # the recursive calls), temporarily remove default value
403 # and put it back.
404 default = opt.pop("default", _NO_SEARCH_DEFAULT_VALUE)
406 # Temporarily replace any env vars so formatter doesn't try to
407 # replace them.
408 value = re.sub(r"\${([^}]+)}", r"<BPSTMP:\1>", value)
409 for name in opt.get("skipNames", {}):
410 value = value.replace(f"{{{name}}}", f"<BPSTMP2:{name}>")
412 # Replace special keys for WMS to fill in.
413 value = re.sub(r"{wms([^}]+)}", lambda x: f"<WMS:{x[1][0].lower() + x[1][1:]}>", value)
415 _LOG.debug("before formatter.format=%s", value)
416 value = self.formatter.format(value, self, opt)
417 _LOG.debug("after formatter.format=%s", value)
419 # Replace any temporary place holders.
420 value = re.sub(r"<BPSTMP:([^>]+)>", r"${\1}", value)
421 value = re.sub(r"<BPSTMP2:([^>]+)>", r"{\1}", value)
423 # if default was originally in opt
424 if default != _NO_SEARCH_DEFAULT_VALUE:
425 opt["default"] = default
427 # check for bpsEval
428 value = re.sub(
429 r"bpsEval\(([^,)]+), ([^)]+)\)", lambda m: str(bps_eval(m.group(1), m.group(2))), value
430 )
431 if "bpsEval" in value:
432 raise ValueError(f"Unparsable bpsEval in '{value}'")
434 return value
436 def modify_value(self, key: str, value: str, opt: dict[str, Any]) -> str:
437 """Modify given value following instructions in given options.
439 Parameters
440 ----------
441 key : `str`
442 Name of given value.
443 value : `str`
444 Value to modify.
445 opt : `dict` [`str`, `~typing.Any`]
446 Options specifying how to modify given value. See :meth: `search`
447 method for list of options.
449 Returns
450 -------
451 new_value : `str`
452 Updated value.
453 """
454 _LOG.debug("modify_value: key=%s, value=%s, opt=%s", key, value, opt)
455 new_value = value
456 if key == "subDirTemplate":
457 # Save if template ends with slash
458 template_endswith_slash = new_value.endswith("/")
460 if opt.get("replaceEnvBps2Shell", False): 460 ↛ 461line 460 didn't jump to line 461 because the condition on line 460 was never true
461 new_value = re.sub(r"<ENV:([^>]+)>", r"${\1}", new_value)
462 elif opt.get("expandEnvVars", True):
463 _LOG.debug("before format=%s", new_value)
464 new_value = re.sub(r"<ENV:([^>]+)>", r"${\1}", new_value)
465 new_value = expandvars(new_value)
466 elif opt.get("replaceEnvShell2Bps", opt.get("replaceEnvVars", False)):
467 # Don't replace double dollar signs or $( to allow
468 # pass-through to WMS
469 new_value = re.sub(r"\${([^$(}]+)}", r"<ENV:\1>", new_value)
470 new_value = re.sub(r"\$([^$(}]+)", r"<ENV:\1>", new_value)
472 _LOG.debug("modify_value: after EnvVars value=%s", new_value)
473 if opt.get("replaceVars", True):
474 new_value = self.replace_vars(new_value, opt)
475 if key == "subDirTemplate":
476 # Make yaml-specified subdirs easier to read
477 # by removing empty subdirs (//). normpath
478 # removes any trailing slash.
479 new_value = normpath(new_value)
480 # Check if subDirTemplate pattern actually ends in slash
481 # If so, the value returned should.
482 if template_endswith_slash:
483 new_value += "/"
484 _LOG.debug("modify_value: after replace_vars value=%s", new_value)
485 return new_value
487 def generate_config(self) -> None:
488 """Update config with values generated by bpsGenerateConfig
489 entries.
490 """
491 _LOG.debug("generate_config before: %s", self)
492 self._recursive_generate_config("", self)
493 _LOG.debug("generate_config after: %s", self)
495 def _recursive_generate_config(self, recursive_key: str, sub_config: Config) -> None:
496 """Update config with values generated by bpsGenerateConfig
497 entries.
499 Parameters
500 ----------
501 recursive_key : `str`
502 Corresponds to a new subconfig in which to search
503 and replace bpsGenerateConfig.
505 sub_config : `lsst.daf.butler.Config`
506 The nested config corresponding to the recursive_key.
508 Raises
509 ------
510 ValueError
511 If bpsGenerateConfig value isn't parseable.
512 ImportError
513 If problems importing bpsGenerateConfig's method.
514 """
515 _LOG.debug("recursive_key = '%s'", recursive_key)
516 genkey = "bpsGenerateConfig" # to make it easier to change
518 # Save to avoid dictionary changed size during iteration error.
519 orig_keys = list(sub_config)
520 for key in orig_keys:
521 value = Config.__getitem__(sub_config, key)
522 _LOG.debug("key = %s, type(value) = %s", key, type(value))
523 if isinstance(value, Config):
524 self._recursive_generate_config(f"{recursive_key}.{key}", value)
525 elif key == genkey:
526 value = self.replace_vars(value, {"searchobj": sub_config})
528 m = re.match(r"(\S+)\((.+)\)", value)
529 if m:
530 results = bps_eval(m.group(1), m.group(2))
531 del sub_config[genkey]
532 sub_config.update(results)
533 if recursive_key:
534 self[recursive_key] = sub_config
535 _LOG.debug("After config = %s", self)
536 else:
537 raise ValueError(f"Unparsable {genkey} value='{value}'")
539 def get_search_opts(self, label: str | None = None) -> dict[str, Any]:
540 """Create base search options given a label.
542 Parameters
543 ----------
544 label : `str` or None, optional
545 If given, label for which to define BpsConfig search options.
547 Returns
548 -------
549 search_opts : `dict` [`str`, `~typing.Any`]
550 Base BpsConfig search options for given label.
551 """
552 search_opts = {"curvals": {}}
553 if label:
554 search_opts["curvals"]["label"] = label
555 if label in self["cluster"]:
556 search_opts["curvals"]["curr_cluster"] = label
557 elif label in self["pipetask"]:
558 search_opts["curvals"]["curr_pipetask"] = label
559 elif label in self:
560 search_opts["searchobj"] = self[label]
562 # Site/cloud can be defined per cluster/pipetask/special job
563 found, val = self.search("computeSite", search_opts)
564 if found:
565 search_opts["curvals"]["curr_site"] = val
566 found, val = self.search("computeCloud", search_opts)
567 if found:
568 search_opts["curvals"]["curr_cloud"] = val
570 return search_opts