Coverage for python/lsst/ctrl/bps/bps_config.py: 98%
226 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 08:59 +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 collections.abc import Mapping
41from os.path import expandvars, normpath
42from typing import Any
44from lsst.daf.butler import Config
45from lsst.resources import ResourcePath
46from lsst.utils import doImport
48from .bps_utils import bps_eval
50_LOG = logging.getLogger(__name__)
52# Using lsst.daf.butler.Config to resolve possible includes.
53BPS_DEFAULTS = Config(ResourcePath("resource://lsst.ctrl.bps/etc/bps_defaults.yaml")).toDict()
55BPS_SEARCH_ORDER = ["bps_cmdline", "payload", "cluster", "pipetask", "site", "cloud", "bps_defined"]
57# Need a string that won't be a valid default value
58# to indicate whether default was defined for search.
59# And None is a valid default value.
60_NO_SEARCH_DEFAULT_VALUE = "__NO_SEARCH_DEFAULT_VALUE__"
63class BpsFormatter(string.Formatter):
64 """String formatter class that allows BPS config search options."""
66 def get_field(self, field_name, args, kwargs):
67 _, val = args[0].search(field_name, opt=args[1])
68 return val, field_name
70 def get_value(self, key, args, kwargs):
71 _, val = args[0].search(key, opt=args[1])
72 return val
75class _ChainedEnvConfig(Config):
76 """Config whose merges chain ``${VAR}`` self-references inside
77 ``environment`` sections.
79 When the value being merged for environment variable ``VAR`` contains
80 ``${VAR}`` and this config already holds a value for that variable, the
81 placeholder is replaced by the existing value instead of the whole value
82 being overwritten. This mirrors shell behavior such as
83 ``PATH="/new/dir:${PATH}"`` and lets several config files pulled in via
84 ``includeConfigs`` each prepend to the same variable. Any ``${VAR}``
85 left in the final value is expanded from the job's shell environment as
86 usual.
87 """
89 def update(self, other: Mapping[str, Any]) -> None: # type: ignore[override]
90 # Work on a copy so the caller's mapping is left untouched.
91 updates = Config(other)
92 for path in updates.nameTuples():
93 if len(path) < 2 or path[-2] != "environment":
94 continue
95 name = path[-1]
96 value = updates[path]
97 if not isinstance(value, str) or f"${{{name}}}" not in value:
98 continue
99 previous = self.get(path)
100 if isinstance(previous, str):
101 updates[path] = value.replace(f"${{{name}}}", previous)
102 super().update(updates)
105class BpsConfig(Config):
106 """Contains the configuration for a BPS submission.
108 Parameters
109 ----------
110 other : `str`, `dict`, `~lsst.daf.butler.Config`, `BpsConfig`
111 Path to a YAML file or a dict/Config/BpsConfig containing configuration
112 to copy.
113 search_order : `list` [`str`], optional
114 Root section names in the order in which they should be searched.
115 defaults : `str`, `dict`, `~lsst.daf.butler.Config`, optional
116 Default settings that will be used to prepopulate the config.
117 If the WMS service default settings are available, they will be added
118 afterwards. WMS settings takes precedence over provided defaults.
119 wms_service_class_fqn : `str`, optional
120 Fully qualified name of the WMS service class to use to get plugin's
121 specific default settings. If `None` (default), the WMS service
122 class provided by
124 1. ``other`` config,
125 2. environmental variable ``BPS_WMS_SERVICE_CLASS``,
126 3. default settings
128 will be used instead. The list above also reflects the priorities
129 if the WMS service class is defined in multiple places. For example,
130 the name of service class found in ``other`` takes precedence over
131 the name of the service class provided by the BPS_SERVICE_CLASS and/or
132 the default settings.
134 Raises
135 ------
136 ValueError
137 Raised if the class cannot be instantiated from the provided object.
138 """
140 def __init__(self, other, search_order=None, defaults=None, wms_service_class_fqn=None):
141 # In BPS config, the same setting can be defined multiple times in
142 # different sections. The sections are search in a pre-defined
143 # order. Hence, a value which is found first effectively overrides
144 # values in later sections, if any. To achieve this goal,
145 # the special methods __getitem__ and __contains__ were redefined to
146 # use a custom search function internally. For this reason we can't
147 # use super().__init__(other) as the super class defines its own
148 # __getitem__ which is utilized during the initialization process (
149 # e.g. in expressions like self[<key>]). However, this function will
150 # be overridden by the one defined here, in the subclass. Instead
151 # we just initialize internal data structures and populate them
152 # using the inherited update() method which does not rely on super
153 # class __getitem__ method.
154 super().__init__()
156 # Read the config with environment chaining so that files pulled in
157 # via includeConfigs can each prepend to the same environment
158 # variable rather than the last file replacing the earlier values.
159 try:
160 other_config = _ChainedEnvConfig(other)
161 except Exception as exc:
162 raise ValueError(f"A BpsConfig could not be loaded from other: {other}") from exc
164 config = _ChainedEnvConfig()
166 # Pre-populate the config with default settings if any were provided
167 # by the caller. Include WMS plugin specific defaults and/or
168 # overrides as well if available.
169 if defaults:
170 config.update(defaults)
172 # If the WMS service class was not specified explicitly by the
173 # caller, try to use the value provided by either:
174 #
175 # 1. 'other' config,
176 # 2. environmental variable BPS_WMS_SERVICE_CLASS,
177 # 3. default settings
178 #
179 # (in decreasing priority).
180 if wms_service_class_fqn is None:
181 wms_service_class_fqn = other_config.get(
182 "wmsServiceClass",
183 os.environ.get("BPS_WMS_SERVICE_CLASS", config.get("wmsServiceClass")),
184 )
185 try:
186 wms_service_class = doImport(wms_service_class_fqn)
187 except TypeError:
188 # Do not die if the WMS service class is still not set.
189 pass
190 else:
191 wms_service = wms_service_class({})
192 wms_defaults = wms_service.defaults
193 if wms_defaults:
194 config.update(wms_defaults)
196 # Set the service class to the one which was defaults was used.
197 config["wmsServiceClass"] = wms_service_class_fqn
199 # Include values and/or apply overrides from 'other' config.
200 config.update(other_config)
201 self.update(config)
203 if isinstance(other, BpsConfig):
204 self.formatter = copy.deepcopy(other.formatter)
205 self.search_order = copy.deepcopy(other.search_order) if search_order is None else search_order
206 else:
207 self.formatter = BpsFormatter()
208 self.search_order = BPS_SEARCH_ORDER if search_order is None else search_order
210 # Make sure search sections exist.
211 for key in self.search_order:
212 if not Config.__contains__(self, key):
213 self[key] = {}
215 def copy(self):
216 """Make a copy of config.
218 Returns
219 -------
220 copy : `lsst.ctrl.bps.BpsConfig`
221 A duplicate of itself.
222 """
223 return BpsConfig(self)
225 def get(self, key, default=""):
226 """Return the value for key if key is in the config, else default.
228 If default is not given, it defaults to an empty string.
230 Parameters
231 ----------
232 key : `str`
233 Key to look for in config.
234 default : `~typing.Any`, optional
235 Default value to return if the key is not in the config.
237 Returns
238 -------
239 val : Any
240 Value from config if found, default otherwise.
242 Notes
243 -----
244 The provided default value (an empty string) was chosen to maintain
245 the internal consistency with other methods of the class.
246 """
247 _, val = self.search(key, opt={"default": default})
248 return val
250 def __getitem__(self, name):
251 """Return the value from the config for the given name.
253 Parameters
254 ----------
255 name : `str`
256 Key to look for in config
258 Returns
259 -------
260 val : `str`, `int`, `lsst.ctrl.bps.BpsConfig`, ...
261 Value from config if found.
262 """
263 _, val = self.search(name, {})
265 return val
267 def __contains__(self, name):
268 """Check whether name is in config.
270 Parameters
271 ----------
272 name : `str`
273 Key to look for in config.
275 Returns
276 -------
277 found : `bool`
278 Whether name was in config or not.
279 """
280 found, _ = self.search(name, {})
281 return found
283 def search(self, key, opt=None):
284 """Search for key using given opt following hierarchy rules.
286 Search hierarchy rules: current values, a given search object, and
287 search order of config sections.
289 Parameters
290 ----------
291 key : `str`
292 Key to look for in config.
293 opt : `dict` [`str`, `~typing.Any`], optional
294 Options dictionary to use while searching. All are optional.
296 ``"curvals"``
297 Means to pass in values for search order key
298 (curr_<sectname>) or variable replacements.
299 (`dict`, optional)
300 ``"default"``
301 Value to return if not found. (`~typing.Any`, optional)
302 ``"replaceEnvVars"``
303 Kept for backward compatibility.
304 See ``replaceEnvShell2Bps``.
305 ``"replaceEnvShell2Bps"``
306 If search result is string, whether to replace environment
307 variables inside it with special placeholder (<ENV:name>).
308 By default set to False. (`bool`)
309 ``"replaceEnvBps2Shell"``
310 If search result is string, whether to replace environment
311 placeholder inside it with shell environment syntax.
312 By default set to False. (`bool`)
313 ``"expandEnvVars"``
314 If search result is string, whether to replace environment
315 placeholder inside it with current environment value.
316 By default set to False. (`bool`)
317 ``"replaceVars"``
318 If search result is string, whether to replace variables
319 inside it. By default set to True. (`bool`)
320 ``"required"``
321 If replacing variables, whether to raise exception if
322 variable is undefined. By default set to False. (`bool`)
324 Returns
325 -------
326 found : `bool`
327 Whether name was in config or not.
328 value : `str`, `int`, `lsst.ctrl.bps.BpsConfig`, ...
329 Value from config if found.
330 """
331 _LOG.debug("search: initial key = '%s', opt = '%s'", key, opt)
333 if opt is None:
334 opt = {}
336 found = False
337 value = ""
339 # start with stored current values
340 curvals = None
341 if Config.__contains__(self, "current"): 341 ↛ 342line 341 didn't jump to line 342 because the condition on line 341 was never true
342 curvals = copy.deepcopy(Config.__getitem__(self, "current"))
343 else:
344 curvals = {}
346 # override with current values passed into function if given
347 if "curvals" in opt:
348 for ckey, cval in list(opt["curvals"].items()):
349 _LOG.debug("using specified curval %s = %s", ckey, cval)
350 curvals[ckey] = cval
352 _LOG.debug("curvals = %s", curvals)
354 # There's a problem with the searchobj being a BpsConfig
355 # and its handling of __getitem__. Until that part of
356 # BpsConfig is rewritten, force the searchobj to a Config.
357 if "searchobj" in opt:
358 opt["searchobj"] = Config(opt["searchobj"])
360 if key in curvals:
361 _LOG.debug("found %s in curvals", key)
362 found = True
363 value = curvals[key]
364 elif "searchobj" in opt and key in opt["searchobj"]:
365 found = True
366 value = opt["searchobj"][key]
367 else:
368 for sect in self.search_order:
369 if Config.__contains__(self, sect): 369 ↛ 384line 369 didn't jump to line 384 because the condition on line 369 was always true
370 _LOG.debug("Searching '%s' section for key '%s'", sect, key)
371 search_sect = Config.__getitem__(self, sect)
372 if "curr_" + sect in curvals:
373 currkey = curvals["curr_" + sect]
374 _LOG.debug("currkey for section %s = %s", sect, currkey)
375 if Config.__contains__(search_sect, currkey):
376 search_sect = Config.__getitem__(search_sect, currkey)
378 _LOG.debug("%s %s", key, search_sect)
379 if Config.__contains__(search_sect, key):
380 found = True
381 value = Config.__getitem__(search_sect, key)
382 break
383 else:
384 _LOG.debug("Missing search section '%s' while searching for '%s'", sect, key)
386 # lastly check root values
387 if not found:
388 _LOG.debug("Searching root section for key '%s'", key)
389 if Config.__contains__(self, key):
390 found = True
391 value = Config.__getitem__(self, key)
392 _LOG.debug("root value='%s'", value)
393 else:
394 _LOG.debug("not in root section for key='%s'", key)
395 _LOG.debug(Config(self))
397 if not found and "default" in opt:
398 value = opt["default"]
399 found = True # ????
401 if not found and opt.get("required", False):
402 print(f"\n\nError: search for {key} failed")
403 print("\tcurrent = ", self.get("current"))
404 print("\topt = ", opt)
405 print("\tcurvals = ", curvals)
406 print("\n\n")
407 raise KeyError(f"Error: Search failed {key}")
409 _LOG.debug("found=%s, value=%s", found, value)
411 _LOG.debug("opt=%s %s", opt, type(opt))
412 if found and isinstance(value, str):
413 value = self.modify_value(key, value, opt)
414 _LOG.debug("after modify_value=%s", value)
416 if found and isinstance(value, Config):
417 value = BpsConfig(value, search_order=[])
419 return found, value
421 def replace_vars(self, value: str, opt: dict[str, Any]) -> str:
422 """Replace variables in string with values except those
423 in opt['skipNames'].
425 Parameters
426 ----------
427 value : `str`
428 Value in which to replace variables.
429 opt : `dict` [`str`, Any]
430 Options to be used when searching and replacing values.
431 In particular "skipNames" lists variable names to
432 not replace.
433 """
434 # default only applies to original search key
435 # Instead of doing deep copies of opt (especially with
436 # the recursive calls), temporarily remove default value
437 # and put it back.
438 default = opt.pop("default", _NO_SEARCH_DEFAULT_VALUE)
440 # Temporarily replace any env vars so formatter doesn't try to
441 # replace them.
442 value = re.sub(r"\${([^}]+)}", r"<BPSTMP:\1>", value)
443 for name in opt.get("skipNames", {}):
444 value = value.replace(f"{{{name}}}", f"<BPSTMP2:{name}>")
446 # Replace special keys for WMS to fill in.
447 value = re.sub(r"{wms([^}]+)}", lambda x: f"<WMS:{x[1][0].lower() + x[1][1:]}>", value)
449 _LOG.debug("before formatter.format=%s", value)
450 value = self.formatter.format(value, self, opt)
451 _LOG.debug("after formatter.format=%s", value)
453 # Replace any temporary place holders.
454 value = re.sub(r"<BPSTMP:([^>]+)>", r"${\1}", value)
455 value = re.sub(r"<BPSTMP2:([^>]+)>", r"{\1}", value)
457 # if default was originally in opt
458 if default != _NO_SEARCH_DEFAULT_VALUE:
459 opt["default"] = default
461 # check for bpsEval
462 value = re.sub(
463 r"bpsEval\(([^,)]+), ([^)]+)\)", lambda m: str(bps_eval(m.group(1), m.group(2))), value
464 )
465 if "bpsEval" in value:
466 raise ValueError(f"Unparsable bpsEval in '{value}'")
468 return value
470 def modify_value(self, key: str, value: str, opt: dict[str, Any]) -> str:
471 """Modify given value following instructions in given options.
473 Parameters
474 ----------
475 key : `str`
476 Name of given value.
477 value : `str`
478 Value to modify.
479 opt : `dict` [`str`, `~typing.Any`]
480 Options specifying how to modify given value. See :meth: `search`
481 method for list of options.
483 Returns
484 -------
485 new_value : `str`
486 Updated value.
487 """
488 _LOG.debug("modify_value: key=%s, value=%s, opt=%s", key, value, opt)
489 new_value = value
490 if key == "subDirTemplate":
491 # Save if template ends with slash
492 template_endswith_slash = new_value.endswith("/")
494 if opt.get("replaceEnvBps2Shell", False):
495 new_value = re.sub(r"<ENV:([^>]+)>", r"${\1}", new_value)
496 elif opt.get("expandEnvVars", True):
497 _LOG.debug("before format=%s", new_value)
498 new_value = re.sub(r"<ENV:([^>]+)>", r"${\1}", new_value)
499 new_value = expandvars(new_value)
500 elif opt.get("replaceEnvShell2Bps", opt.get("replaceEnvVars", False)):
501 # Don't replace double dollar signs or $( to allow
502 # pass-through to WMS
503 new_value = re.sub(r"\${([^$(}]+)}", r"<ENV:\1>", new_value)
504 new_value = re.sub(r"\$([^$(}]+)", r"<ENV:\1>", new_value)
506 _LOG.debug("modify_value: after EnvVars value=%s", new_value)
507 if opt.get("replaceVars", True):
508 new_value = self.replace_vars(new_value, opt)
509 if key == "subDirTemplate":
510 # Make yaml-specified subdirs easier to read
511 # by removing empty subdirs (//). normpath
512 # removes any trailing slash.
513 new_value = normpath(new_value)
514 # Check if subDirTemplate pattern actually ends in slash
515 # If so, the value returned should.
516 if template_endswith_slash:
517 new_value += "/"
518 _LOG.debug("modify_value: after replace_vars value=%s", new_value)
519 return new_value
521 def generate_config(self) -> None:
522 """Update config with values generated by bpsGenerateConfig
523 entries.
524 """
525 _LOG.debug("generate_config before: %s", self)
526 self._recursive_generate_config("", self)
527 _LOG.debug("generate_config after: %s", self)
529 def _recursive_generate_config(self, recursive_key: str, sub_config: Config) -> None:
530 """Update config with values generated by bpsGenerateConfig
531 entries.
533 Parameters
534 ----------
535 recursive_key : `str`
536 Corresponds to a new subconfig in which to search
537 and replace bpsGenerateConfig.
539 sub_config : `lsst.daf.butler.Config`
540 The nested config corresponding to the recursive_key.
542 Raises
543 ------
544 ValueError
545 If bpsGenerateConfig value isn't parseable.
546 ImportError
547 If problems importing bpsGenerateConfig's method.
548 """
549 _LOG.debug("recursive_key = '%s'", recursive_key)
550 genkey = "bpsGenerateConfig" # to make it easier to change
552 # Save to avoid dictionary changed size during iteration error.
553 orig_keys = list(sub_config)
554 for key in orig_keys:
555 value = Config.__getitem__(sub_config, key)
556 _LOG.debug("key = %s, type(value) = %s", key, type(value))
557 if isinstance(value, Config):
558 self._recursive_generate_config(f"{recursive_key}.{key}", value)
559 elif key == genkey:
560 value = self.replace_vars(value, {"searchobj": sub_config})
562 m = re.match(r"(\S+)\((.+)\)", value)
563 if m:
564 results = bps_eval(m.group(1), m.group(2))
565 del sub_config[genkey]
566 sub_config.update(results)
567 if recursive_key:
568 self[recursive_key] = sub_config
569 _LOG.debug("After config = %s", self)
570 else:
571 raise ValueError(f"Unparsable {genkey} value='{value}'")
573 def get_search_opts(self, label: str | None = None) -> dict[str, Any]:
574 """Create base search options given a label.
576 Parameters
577 ----------
578 label : `str` or None, optional
579 If given, label for which to define BpsConfig search options.
581 Returns
582 -------
583 search_opts : `dict` [`str`, `~typing.Any`]
584 Base BpsConfig search options for given label.
585 """
586 search_opts = {"curvals": {}}
587 if label:
588 search_opts["curvals"]["label"] = label
589 if label in self["cluster"]:
590 search_opts["curvals"]["curr_cluster"] = label
591 elif label in self["pipetask"]:
592 search_opts["curvals"]["curr_pipetask"] = label
593 elif label in self:
594 search_opts["searchobj"] = self[label]
596 # Site/cloud can be defined per cluster/pipetask/special job
597 found, val = self.search("computeSite", search_opts)
598 if found:
599 search_opts["curvals"]["curr_site"] = val
600 found, val = self.search("computeCloud", search_opts)
601 if found:
602 search_opts["curvals"]["curr_cloud"] = val
604 return search_opts