Coverage for python/lsst/ctrl/mpexec/cli/cmd/commands.py: 88%
187 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-25 08:40 +0000
1# This file is part of ctrl_mpexec.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://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 <http://www.gnu.org/licenses/>.
28import sys
29from collections.abc import Iterator, Sequence
30from contextlib import contextmanager
31from functools import partial
32from importlib import import_module
33from tempfile import NamedTemporaryFile
34from typing import Any
36import click
38from lsst.ctrl.mpexec.showInfo import ShowInfo
39from lsst.daf.butler.cli.opt import (
40 collections_option,
41 confirm_option,
42 options_file_option,
43 processes_option,
44 repo_argument,
45 where_option,
46)
47from lsst.daf.butler.cli.utils import catch_and_exit, option_section, unwrap
48from lsst.pipe.base.quantum_reports import Report
49from lsst.utils.threads import disable_implicit_threading
51from .. import opt as ctrlMpExecOpts
52from .. import script
53from ..script import confirmable
54from ..utils import PipetaskCommand, collect_pipeline_actions
56epilog = unwrap(
57 """Notes:
59--task, --delete, --config, --config-file, and --instrument action options can
60appear multiple times; all values are used, in order left to right.
62FILE reads command-line options from the specified file. Data may be
63distributed among multiple lines (e.g. one option per line). Data after # is
64treated as a comment and ignored. Blank lines and lines starting with # are
65ignored.)
66"""
67)
70def _unhandledShow(show: ShowInfo, cmd: str) -> None:
71 if show.unhandled: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true
72 print(
73 f"The following '--show' options were not known to the {cmd} command: "
74 f"{', '.join(show.unhandled)}",
75 file=sys.stderr,
76 )
79@click.command(cls=PipetaskCommand, epilog=epilog, short_help="Build pipeline definition.")
80@click.pass_context
81@ctrlMpExecOpts.show_option()
82@ctrlMpExecOpts.pipeline_build_options()
83@option_section(sectionText="")
84@options_file_option()
85@catch_and_exit
86def build(ctx: click.Context, **kwargs: Any) -> None:
87 """Build and optionally save pipeline definition.
89 This does not require input data to be specified.
90 """
91 kwargs = collect_pipeline_actions(ctx, **kwargs)
92 show = ShowInfo(kwargs.pop("show", []))
93 if kwargs.get("butler_config") is not None and ( 93 ↛ 96line 93 didn't jump to line 96 because the condition on line 93 was never true
94 {"pipeline-graph", "task-graph"}.isdisjoint(show.commands) and not kwargs.get("pipeline_dot")
95 ):
96 raise click.ClickException(
97 "--butler-config was provided but nothing uses it "
98 "(only --show pipeline-graph, --show task-graph and --pipeline-dot do)."
99 )
100 script.build(**kwargs, show=show)
101 _unhandledShow(show, "build")
104@contextmanager
105def coverage_context(kwargs: dict[str, Any]) -> Iterator[None]:
106 """Enable coverage recording."""
107 packages = kwargs.pop("cov_packages", ())
108 report = kwargs.pop("cov_report", True)
109 if not kwargs.pop("coverage", False):
110 yield
111 return
112 # Lazily import coverage only when we might need it
113 try:
114 coverage = import_module("coverage")
115 except ModuleNotFoundError:
116 raise click.ClickException("coverage was requested but the coverage package is not installed.")
117 with NamedTemporaryFile("w") as rcfile:
118 rcfile.write(
119 """
120[run]
121branch = True
122concurrency = multiprocessing
123"""
124 )
125 if packages: 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true
126 packages_str = ",".join(packages)
127 rcfile.write(f"source_pkgs = {packages_str}\n")
128 rcfile.flush()
129 cov = coverage.Coverage(config_file=rcfile.name)
130 cov.start()
131 try:
132 yield
133 finally:
134 cov.stop()
135 cov.save()
136 if report: 136 ↛ exitline 136 didn't jump to the function exit
137 outdir = "./covhtml"
138 cov.html_report(directory=outdir)
139 click.echo(f"Coverage report written to {outdir}.")
142@click.command(cls=PipetaskCommand, epilog=epilog)
143@click.pass_context
144@ctrlMpExecOpts.show_option()
145@ctrlMpExecOpts.pipeline_build_options(skip_butler_config=True)
146@ctrlMpExecOpts.qgraph_options()
147@ctrlMpExecOpts.butler_options()
148@option_section(sectionText="")
149@options_file_option()
150@catch_and_exit
151def qgraph(ctx: click.Context, **kwargs: Any) -> None:
152 """Build and optionally save quantum graph."""
153 kwargs = collect_pipeline_actions(ctx, **kwargs)
154 summary = kwargs.pop("summary", None)
155 with coverage_context(kwargs):
156 show = ShowInfo(kwargs.pop("show", []))
157 # The only reason 'build' might want a butler is to resolve the
158 # pipeline graph for its own 'show' options, which wouldn't run in this
159 # context. Take it out of the kwargs so it doesn't instantiate a
160 # butler unnecessarily.
161 butler_config = kwargs.pop("butler_config", None)
162 pipeline_graph_factory = script.build(**kwargs, show=show)
163 kwargs["butler_config"] = butler_config
164 if show.handled and not show.unhandled: 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true
165 print(
166 "No quantum graph generated. The --show option was given and all options were processed.",
167 file=sys.stderr,
168 )
169 return
170 if ( 170 ↛ 180line 170 didn't jump to line 180 because the condition on line 170 was never true
171 qgraph := script.qgraph(
172 pipeline_graph_factory,
173 **kwargs,
174 show=show,
175 # Making a summary report requires that we load the same graph
176 # components as execution.
177 for_execution=(summary is not None),
178 )
179 ) is None:
180 raise click.ClickException("QuantumGraph was empty; ERROR logs above should provide details.")
181 # QuantumGraph-only summary call here since script.qgraph also called
182 # by run methods.
183 if summary:
184 report = Report(qgraphSummary=qgraph._make_summary())
185 with open(summary, "w") as out:
186 # Do not save fields that are not set.
187 out.write(report.model_dump_json(exclude_none=True, indent=2))
189 _unhandledShow(show, "qgraph")
192@click.command(cls=PipetaskCommand, epilog=epilog)
193@ctrlMpExecOpts.run_options()
194@catch_and_exit
195def run(ctx: click.Context, **kwargs: Any) -> None:
196 """Build and execute pipeline and quantum graph."""
197 kwargs = collect_pipeline_actions(ctx, **kwargs)
198 if not kwargs.get("enable_implicit_threading"):
199 # Limit thread pools created during pipeline construction and graph
200 # generation, not just those created during execution. script.run
201 # repeats this for callers that do not go through this command.
202 disable_implicit_threading()
203 with coverage_context(kwargs):
204 show = ShowInfo(kwargs.pop("show", []))
205 pipeline_graph_factory = script.build(**kwargs, show=show)
206 if show.handled and not show.unhandled: 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 print(
208 "No quantum graph generated or pipeline executed. "
209 "The --show option was given and all options were processed.",
210 file=sys.stderr,
211 )
212 return
213 if (qgraph := script.qgraph(pipeline_graph_factory, for_execution=True, **kwargs, show=show)) is None: 213 ↛ 214line 213 didn't jump to line 214 because the condition on line 213 was never true
214 raise click.ClickException("QuantumGraph was empty; ERROR logs above should provide details.")
215 _unhandledShow(show, "run")
216 if show.handled: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 print(
218 "No pipeline executed. The --show option was given and all options were processed.",
219 file=sys.stderr,
220 )
221 return
222 script.run(qgraph, **kwargs)
225@click.command(cls=PipetaskCommand)
226@ctrlMpExecOpts.butler_config_option()
227@ctrlMpExecOpts.collection_argument()
228@confirm_option()
229@ctrlMpExecOpts.recursive_option(
230 help="""If the parent CHAINED collection has child CHAINED collections,
231 search the children until nested chains that start with the parent's name
232 are removed."""
233)
234def purge(confirm: bool, **kwargs: Any) -> None:
235 """Remove a CHAINED collection and its contained collections.
237 COLLECTION is the name of the chained collection to purge. it must not be a
238 child of any other CHAINED collections
240 Child collections must be members of exactly one collection.
242 The collections that will be removed will be printed, there will be an
243 option to continue or abort (unless using --no-confirm).
244 """
245 confirmable.confirm(partial(script.purge, **kwargs), confirm)
248@click.command(cls=PipetaskCommand)
249@ctrlMpExecOpts.butler_config_option()
250@ctrlMpExecOpts.collection_argument()
251@confirm_option()
252def cleanup(confirm: bool, **kwargs: Any) -> None:
253 """Remove non-members of CHAINED collections.
255 Removes collections that start with the same name as a CHAINED
256 collection but are not members of that collection.
257 """
258 confirmable.confirm(partial(script.cleanup, **kwargs), confirm)
261@click.command(cls=PipetaskCommand)
262@repo_argument()
263@ctrlMpExecOpts.qgraph_argument()
264@ctrlMpExecOpts.config_search_path_option()
265@ctrlMpExecOpts.qgraph_id_option()
266@ctrlMpExecOpts.coverage_options()
267def pre_exec_init_qbb(repo: str, qgraph: str, **kwargs: Any) -> None:
268 """Execute pre-exec-init on Quantum-Backed Butler.
270 REPO is the location of the butler/registry config file.
272 QGRAPH is the path to a serialized Quantum Graph file.
273 """
274 with coverage_context(kwargs):
275 script.pre_exec_init_qbb(repo, qgraph, **kwargs)
278@click.command(cls=PipetaskCommand)
279@repo_argument()
280@ctrlMpExecOpts.qgraph_argument()
281@ctrlMpExecOpts.config_search_path_option()
282@ctrlMpExecOpts.qgraph_id_option()
283@ctrlMpExecOpts.qgraph_node_id_option()
284@processes_option()
285@ctrlMpExecOpts.pdb_option()
286@ctrlMpExecOpts.profile_option()
287@ctrlMpExecOpts.coverage_options()
288@ctrlMpExecOpts.debug_option()
289@ctrlMpExecOpts.start_method_option()
290@ctrlMpExecOpts.timeout_option()
291@ctrlMpExecOpts.fail_fast_option()
292@ctrlMpExecOpts.raise_on_partial_outputs_option()
293@ctrlMpExecOpts.summary_option()
294@ctrlMpExecOpts.enable_implicit_threading_option()
295@ctrlMpExecOpts.cores_per_quantum_option()
296@ctrlMpExecOpts.memory_per_quantum_option()
297@ctrlMpExecOpts.no_existing_outputs_option()
298def run_qbb(repo: str, qgraph: str, **kwargs: Any) -> None:
299 """Execute pipeline using Quantum-Backed Butler.
301 REPO is the location of the butler/registry config file.
303 QGRAPH is the path to a serialized Quantum Graph file.
304 """
305 with coverage_context(kwargs):
306 script.run_qbb(butler_config=repo, qgraph=qgraph, **kwargs)
309@click.command(cls=PipetaskCommand)
310@ctrlMpExecOpts.qgraph_argument()
311@ctrlMpExecOpts.run_argument()
312@ctrlMpExecOpts.output_qgraph_argument()
313@ctrlMpExecOpts.metadata_run_key_option()
314@ctrlMpExecOpts.update_graph_id_option()
315def update_graph_run(
316 qgraph: str,
317 run: str,
318 output_qgraph: str,
319 metadata_run_key: str,
320 update_graph_id: bool,
321) -> None:
322 """Update existing quantum graph with new output run name and re-generate
323 output dataset IDs.
325 QGRAPH is the URL to a serialized Quantum Graph file.
327 RUN is the new RUN collection name for output graph.
329 OUTPUT_QGRAPH is the URL to store the updated Quantum Graph.
330 """
331 script.update_graph_run(qgraph, run, output_qgraph, metadata_run_key, update_graph_id)
334@click.command(cls=PipetaskCommand)
335@repo_argument()
336@click.argument("qgraphs", nargs=-1)
337@collections_option()
338@where_option()
339@click.option(
340 "--full-output-filename",
341 default="",
342 help="Output report as a file with this name. "
343 "For pipetask report on one graph, this should be a yaml file. For multiple graphs "
344 "or when using the --force-v2 option, this should be a json file. We will be "
345 "deprecating the single-graph-only (QuantumGraphExecutionReport) option soon.",
346)
347@click.option("--logs/--no-logs", default=True, help="Get butler log datasets for extra information.")
348@click.option(
349 "--brief",
350 default=False,
351 is_flag=True,
352 help="Only show counts in report (a brief summary). Note that counts are"
353 " also printed to the screen when using the --full-output-filename option.",
354)
355@click.option(
356 "--curse-failed-logs",
357 is_flag=True,
358 default=False,
359 help="If log datasets are missing in v2 (QuantumProvenanceGraph), mark them as cursed",
360)
361@click.option(
362 "--force-v2",
363 is_flag=True,
364 default=False,
365 help="Use the QuantumProvenanceGraph instead of the QuantumGraphExecutionReport, "
366 "even when there is only one qgraph. Otherwise, the QuantumGraphExecutionReport "
367 "will run on one graph by default.",
368)
369@click.option(
370 "--read-caveats",
371 type=click.Choice(["exhaustive", "lazy", "none"], case_sensitive=False),
372 default="lazy",
373)
374@click.option(
375 "--use-qbb/--no-use-qbb",
376 is_flag=True,
377 default=True,
378 help="Whether to use a quantum-backed butler for metadata and log reads.",
379)
380@click.option(
381 "--view-graph",
382 is_flag=True,
383 default=False,
384 help="Display pipeline processing status as a graph on stdout instead of a plain-text summary.",
385)
386@processes_option()
387def report(
388 repo: str,
389 qgraphs: Sequence[str],
390 collections: Sequence[str] | None,
391 where: str,
392 full_output_filename: str = "",
393 logs: bool = True,
394 brief: bool = False,
395 curse_failed_logs: bool = False,
396 force_v2: bool = False,
397 read_caveats: str = "lazy",
398 use_qbb: bool = True,
399 processes: int = 1,
400 view_graph: bool = False,
401) -> None:
402 """Summarize the state of executed quantum graph(s), with counts of failed,
403 successful and expected quanta, as well as counts of output datasets and
404 their query (visible/shadowed) states. Analyze one or more attempts at the
405 same processing on the same dataquery-identified "group" and resolve
406 recoveries and persistent failures. Identify mismatch errors between
407 attempts.
409 Save the report as a file (``--full-output-filename``) or print it to
410 stdout (default). If the terminal is overwhelmed with data_ids from
411 failures try the ``--brief`` option.
413 Butler ``collections`` and ``where`` options are for use in
414 `lsst.daf.butler.Registry.queryDatasets` if paring down the collections
415 would be useful. Pass collections in order of most to least recent. By
416 default the collections and query will be taken from the graphs.
418 REPO is the location of the butler/registry config file.
420 QGRAPHS is a sequence of links to serialized Quantum Graphs which have
421 been executed and are to be analyzed. Pass the graphs in order of first to
422 last executed.
423 """
424 if any([force_v2, len(qgraphs) > 1, collections, where, curse_failed_logs]):
425 script.report_v2(
426 repo,
427 qgraphs,
428 collections,
429 where,
430 full_output_filename,
431 logs,
432 brief,
433 curse_failed_logs,
434 read_caveats=(read_caveats if read_caveats != "none" else None), # type: ignore[arg-type]
435 use_qbb=use_qbb,
436 n_cores=processes,
437 view_graph=view_graph,
438 )
439 else:
440 assert len(qgraphs) == 1, "Cannot make a report without a quantum graph."
441 script.report(repo, qgraphs[0], full_output_filename, logs, brief)
444@click.command(cls=PipetaskCommand)
445@click.argument("filenames", nargs=-1)
446@click.option(
447 "--full-output-filename",
448 default="",
449 help="Output report as a file with this name (json).",
450)
451@click.option(
452 "--brief",
453 default=False,
454 is_flag=True,
455 help="Only show counts in report (a brief summary). Note that counts are"
456 " also printed to the screen when using the --full-output-filename option.",
457)
458def aggregate_reports(
459 filenames: Sequence[str], full_output_filename: str | None, brief: bool = False
460) -> None:
461 """Aggregate pipetask report output on disjoint data-id groups into one
462 Summary over common tasks and datasets. Intended for use when the same
463 pipeline has been run over all groups (i.e., to aggregate all reports
464 for a given step). This functionality is only compatible with reports
465 from the `~lsst.pipe.base.quantum_provenance_graph.QuantumProvenanceGraph`,
466 so the reports must be run over multiple groups or with the ``--force-v2``
467 option.
469 Save the report as a file (``--full-output-filename``) or print it to
470 stdout (default). If the terminal is overwhelmed with data_ids from
471 failures try the ``--brief`` option.
473 FILENAMES are the space-separated paths to json file output created by
474 pipetask report.
475 """
476 script.aggregate_reports(filenames, full_output_filename, brief)