Coverage for python / lsst / ctrl / bps / initialize.py: 19%

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/>. 

27 

28"""Driver for initializing a BPS submission.""" 

29 

30__all__ = [ 

31 "custom_job_validator", 

32 "init_submission", 

33 "out_collection_validator", 

34 "output_run_validator", 

35 "submit_path_validator", 

36] 

37 

38import getpass 

39import logging 

40import re 

41import shutil 

42from collections.abc import Callable, Iterable 

43 

44from lsst.ctrl.bps import BPS_DEFAULTS, BPS_SEARCH_ORDER, BpsConfig 

45from lsst.ctrl.bps.bps_utils import _dump_env_info, _dump_pkg_info, mkdir 

46from lsst.pipe.base import Instrument 

47from lsst.utils import doImport 

48 

49_LOG = logging.getLogger(__name__) 

50 

51 

52def init_submission( 

53 config_file: str, validators: Iterable[Callable[[BpsConfig], None]] = (), **kwargs 

54) -> BpsConfig: 

55 """Initialize BPS configuration and create submit directory. 

56 

57 Parameters 

58 ---------- 

59 config_file : `str` 

60 Name of the configuration file. 

61 validators : `Iterable[Callable[[BpsConfig], None]]`, optional 

62 A list of functions performing checks on the given configuration. 

63 Each function should take a single argument, a BpsConfig object, and 

64 raise if the check fails. By default, no checks are performed. 

65 **kwargs : `~typing.Any` 

66 Additional modifiers to the configuration. 

67 

68 Returns 

69 ------- 

70 config : `lsst.ctrl.bps.BpsConfig` 

71 Batch Processing Service configuration. 

72 """ 

73 config = BpsConfig( 

74 config_file, 

75 search_order=BPS_SEARCH_ORDER, 

76 defaults=BPS_DEFAULTS, 

77 wms_service_class_fqn=kwargs.get("wms_service"), 

78 ) 

79 

80 # Override config with command-line values. 

81 # Handle diffs between pipetask argument names vs bps yaml 

82 translation = { 

83 "input": "inCollection", 

84 "output_run": "outputRun", 

85 "qgraph": "qgraphFile", 

86 "pipeline": "pipelineYaml", 

87 "wms_service": "wmsServiceClass", 

88 "compute_site": "computeSite", 

89 } 

90 for key, value in kwargs.items(): 

91 # Don't want to override config with None or empty string values. 

92 if value: 

93 # pipetask argument parser converts some values to list, 

94 # but bps will want string. 

95 if not isinstance(value, str) and isinstance(value, Iterable): 

96 value = ",".join(value) 

97 new_key = translation.get(key, re.sub(r"_(\S)", lambda match: match.group(1).upper(), key)) 

98 config[f".bps_cmdline.{new_key}"] = value 

99 

100 # Run validation tests on the given config if any. 

101 for validator in validators: 

102 validator(config) 

103 

104 # Set some initial values 

105 config[".bps_defined.timestamp"] = Instrument.makeCollectionTimestamp() 

106 

107 if "operator" not in config: 

108 config[".bps_defined.operator"] = getpass.getuser() 

109 

110 if "uniqProcName" not in config: 

111 config[".bps_defined.uniqProcName"] = config["outputRun"].replace("/", "_") 

112 

113 # If requested, run WMS plugin checks early in submission process to 

114 # ensure WMS has what it will need for prepare() or submit(). 

115 if kwargs.get("runWmsSubmissionChecks", False): 

116 found, wms_class = config.search("wmsServiceClass") 

117 if not found: 

118 raise KeyError("Missing wmsServiceClass in bps config. Aborting.") 

119 

120 # Check that can import wms service class. 

121 wms_service_class = doImport(wms_class) 

122 wms_service = wms_service_class(config) 

123 

124 try: 

125 wms_service.run_submission_checks() 

126 except NotImplementedError: 

127 # Allow various plugins to implement only when needed to do extra 

128 # checks. 

129 _LOG.debug("run_submission_checks is not implemented in %s.", wms_class) 

130 else: 

131 _LOG.debug("Skipping submission checks.") 

132 

133 # Replace all bpsGenerateConfig 

134 config.generate_config() 

135 

136 # Make submit directory to contain all outputs. 

137 submit_path = mkdir(config["submitPath"]) 

138 config[".bps_defined.submitPath"] = str(submit_path) 

139 

140 # save copy of configs (orig and expanded config) 

141 shutil.copy2(config_file, submit_path) 

142 with open(f"{submit_path}/{config['uniqProcName']}_config.yaml", "w") as fh: 

143 config.dump(fh) 

144 

145 # Dump information about runtime environment and software versions in use. 

146 _dump_env_info(f"{submit_path}/{config['uniqProcName']}.env.info.yaml") 

147 _dump_pkg_info(f"{submit_path}/{config['uniqProcName']}.pkg.info.yaml") 

148 

149 return config 

150 

151 

152def output_run_validator(config: BpsConfig) -> None: 

153 """Check if 'outputRun' is specified in BPS config. 

154 

155 Parameters 

156 ---------- 

157 config : `BpsConfig` 

158 BPS configuration that needs to be validated. 

159 

160 Raises 

161 ------ 

162 KeyError 

163 Raised if 'outputRun' is not specified in the BPS configuration. 

164 """ 

165 if "outputRun" not in config: 

166 raise KeyError("Must specify the output run collection using 'outputRun'") 

167 

168 

169def submit_path_validator(config: BpsConfig) -> None: 

170 """Check if 'submitPath' is specified in BPS config. 

171 

172 Parameters 

173 ---------- 

174 config : `BpsConfig` 

175 BPS configuration that needs to be validated. 

176 

177 Raises 

178 ------ 

179 KeyError 

180 Raised if 'submitPath' is not specified in the BPS configuration. 

181 """ 

182 if "submitPath" not in config: 

183 raise KeyError("Must specify the submit-side run directory using 'submitPath'") 

184 

185 

186def out_collection_validator(config: BpsConfig) -> None: 

187 """Check if 'outCollection' is *not* specified in BPS config. 

188 

189 Parameters 

190 ---------- 

191 config : `BpsConfig` 

192 BPS configuration that needs to be validated. 

193 

194 Raises 

195 ------ 

196 KeyError 

197 Raised if 'outCollection' *is* specified in the BPS configuration. 

198 """ 

199 if "outCollection" in config: 

200 raise KeyError("'outCollection' is deprecated. Replace all references to it with 'outputRun'.") 

201 

202 

203def custom_job_validator(config: BpsConfig) -> None: 

204 """Check if 'customJob' is specified in BPS config. 

205 

206 Parameters 

207 ---------- 

208 config : `BpsConfig` 

209 BPS configuration that needs to be validated. 

210 

211 Raises 

212 ------ 

213 KeyError 

214 Raised if 'customJob' is not specified in the BPS configuration. 

215 """ 

216 if "customJob" not in config and "executable" not in config["customJob"]: 

217 raise KeyError("Must specify the details of the script to execute using 'customJob'.")