Coverage for python / lsst / ctrl / bps / construct.py: 13%
78 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-20 01:43 -0800
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"""Driver for constructing a generic workflow running a custom job."""
30__all__ = ["construct"]
32import logging
33import shutil
34from collections.abc import Callable
35from pathlib import Path
37from lsst.ctrl.bps import (
38 BpsConfig,
39 GenericWorkflow,
40 GenericWorkflowExec,
41 GenericWorkflowFile,
42 GenericWorkflowJob,
43)
44from lsst.ctrl.bps.transform import _get_job_values
46_LOG = logging.getLogger(__name__)
49def construct(config: BpsConfig) -> tuple[GenericWorkflow, BpsConfig]:
50 """Create a workflow for running a custom job.
52 Parameters
53 ----------
54 config : `lsst.ctrl.bps.BpsConfig`
55 Configuration values to be used by submission.
57 Returns
58 -------
59 generic_workflow : `lsst.ctrl.bps.GenericWorkflow`
60 Generic workflow for running a standalone job.
61 generic_workflow_config : `lsst.ctrl.BpsConfig`
62 Configuration to accompany created generic workflow.
63 """
64 generic_workflow, generic_workflow_config = create_custom_workflow(config)
65 return generic_workflow, generic_workflow_config
68def create_custom_workflow(config: BpsConfig) -> tuple[GenericWorkflow, BpsConfig]:
69 """Create a workflow that will run a custom job.
71 Parameters
72 ----------
73 config : `lsst.ctrl.bps.BpsConfig`
74 BPS configuration.
76 Returns
77 -------
78 generic_workflow : `lsst.ctrl.bps.GenericWorkflow`
79 Generic workflow for running a custom job.
80 generic_workflow_config : `lsst.ctrl.BpsConfig`
81 Configuration to accompany created generic workflow.
82 """
83 gwjob, inputs, outputs = create_custom_job(config)
85 _, name = config.search("uniqProcName", opt={"required": True})
86 generic_workflow = GenericWorkflow(name)
87 generic_workflow.add_job(gwjob)
88 generic_workflow.run_attrs.update(
89 {
90 "bps_isjob": "True",
91 "bps_iscustom": "True",
92 "bps_project": config["project"],
93 "bps_campaign": config["campaign"],
94 "bps_run": generic_workflow.name,
95 "bps_operator": config["operator"],
96 "bps_payload": config["payloadName"],
97 "bps_runsite": config["computeSite"],
98 }
99 )
100 if inputs:
101 generic_workflow.add_job_inputs(gwjob.name, inputs)
102 if outputs:
103 generic_workflow.add_job_outputs(gwjob.name, outputs)
105 generic_workflow_config = BpsConfig(config)
106 generic_workflow_config["workflowName"] = config["uniqProcName"]
107 generic_workflow_config["workflowPath"] = config["submitPath"]
109 return generic_workflow, generic_workflow_config
112def create_custom_job(
113 config: BpsConfig,
114) -> tuple[GenericWorkflowJob, list[GenericWorkflowFile], list[GenericWorkflowFile]]:
115 """Create a job that will run a custom command or script.
117 Parameters
118 ----------
119 config : `lsst.ctrl.bps.BpsConfig`
120 BPS configuration.
122 Returns
123 -------
124 job : `lsst.ctrl.bps.GenericWorkflowJob`
125 A custom job responsible for running the command.
126 inputs : `list` [`lsst.ctrl.bps.GenericWorkflowFile`]
127 List of job's input files, empty if the job has no input files.
128 outputs : `list` [`lsst.ctrl.bps.GenericWorkflowFile`]
129 List of job's output files, empty if the job has no output files.
130 """
131 prefix = Path(config["submitPath"])
132 job_label = "customJob"
134 search_opts = {"searchobj": config[job_label], "curvals": {}}
135 found, value = config.search("computeSite", opt=search_opts)
136 if found:
137 search_opts["curvals"]["curr_site"] = value
138 found, value = config.search("computeCloud", opt=search_opts)
139 if found:
140 search_opts["curvals"]["curr_cloud"] = value
142 script_file = Path(config[f".{job_label}.executable"])
143 script_name = script_file.name
145 shutil.copy2(script_file, prefix)
147 job = GenericWorkflowJob(name=script_name, label=job_label)
148 job_values = _get_job_values(config, search_opts, "")
149 for attr, value in job_values.items():
150 if not getattr(job, attr):
151 setattr(job, attr, value)
152 job.executable = GenericWorkflowExec(
153 name=script_name, src_uri=str(prefix / script_name), transfer_executable=True
154 )
155 _, job.arguments = config.search("arguments", opt=search_opts | {"replaceVars": False})
157 inputs = []
158 found, mapping = config.search("inputs", opt=search_opts)
159 if found:
160 inputs = create_job_files(mapping, prefix, path_creator=create_input_path)
162 outputs = []
163 found, mapping = config.search("outputs", opt=search_opts)
164 if found:
165 outputs = create_job_files(mapping, prefix, path_creator=create_output_path)
167 for gwfile in inputs + outputs:
168 job.arguments = job.arguments.replace(f"{{{gwfile.name}}}", f"<FILE:{gwfile.name}>")
170 return job, inputs, outputs
173def create_job_files(
174 file_specs: BpsConfig, prefix: str | Path, path_creator: Callable[[Path, Path], Path]
175) -> list[GenericWorkflowFile]:
176 """Create files for a job.
178 Parameters
179 ----------
180 file_specs : `lsst.ctrl.bps.BpsConfig`
181 The mapping between file keys and file paths.
182 prefix : `str` | `pathlib.Path`
183 The root directory to which the files will be written.
184 path_creator : `Callable` [[`Path`, `Path`], `Path`]
185 File category that determines actions that need to be taken during
186 file creation.
188 Returns
189 -------
190 gwfiles : `list` [`lsst.ctrl.bps.GenericWorkflowFile`]
191 List of files created for the job.
192 """
193 prefix = Path(prefix)
195 gwfiles = []
196 for key, path in file_specs.items():
197 src = Path(path)
198 dest = path_creator(src, prefix)
199 gwfiles.append(GenericWorkflowFile(name=key, src_uri=str(dest), wms_transfer=True))
200 return gwfiles
203def create_input_path(path: Path, prefix: Path) -> Path:
204 """Process an input path.
206 Parameters
207 ----------
208 path : `pathlib.Path`
209 The input path.
210 prefix : `pathlib.Path`
211 The root directory to which the file will be written.
213 Raises
214 ------
215 ValueError
216 Raised if the input path does not exist or is a directory.
217 """
218 if path.exists():
219 if path.is_dir():
220 raise ValueError(f"input path '{path} is a directory, must be file")
221 else:
222 raise ValueError(f"input path '{path}' does not exist")
223 dest = prefix / path.name
224 shutil.copy2(path, dest)
225 return dest
228def create_output_path(path: Path, prefix: Path) -> Path:
229 """Process an output path.
231 Parameters
232 ----------
233 path : `pathlib.Path`
234 The output path.
235 prefix : `pathlib.Path`
236 The root directory to which the file will be written.
237 """
238 if path.is_absolute():
239 dest = path
240 else:
241 dest = prefix / path
242 dest.parent.mkdir(parents=True, exist_ok=True)
243 return dest