Coverage for python / lsst / ctrl / mpexec / cli / script / report.py: 15%
62 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-15 06:33 -0300
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/>.
27import pprint
28from collections.abc import Sequence
29from typing import Literal
31from astropy.table import Table
33from lsst.daf.butler import Butler
34from lsst.pipe.base import QuantumGraph
35from lsst.pipe.base.execution_reports import QuantumGraphExecutionReport
36from lsst.pipe.base.quantum_provenance_graph import QuantumProvenanceGraph, Summary
39def report(
40 butler_config: str,
41 qgraph_uri: str,
42 full_output_filename: str | None,
43 logs: bool = True,
44 brief: bool = False,
45) -> None:
46 """Summarize the produced, missing and expected quanta and
47 datasets belonging to an executed quantum graph using the
48 `QuantumGraphExecutionReport`.
50 Parameters
51 ----------
52 butler_config : `str`
53 The Butler used for this report. This should match the Butler used for
54 the run associated with the executed quantum graph.
55 qgraph_uri : `str`
56 The uri of the location of said quantum graph.
57 full_output_filename : `str`
58 Output the full summary report to a yaml file (named herein). Each data
59 id and error message is keyed to a quantum graph node id. A convenient
60 output format for error-matching and cataloguing tools such as the ones
61 in the Campaign Management database. If this is not included, quanta
62 and dataset information will be printed to the command-line instead.
63 logs : `bool`
64 Get butler log datasets for extra information (error messages).
65 brief : `bool`
66 List only the counts (or data_ids if number of failures < 5). This
67 option is good for those who just want to see totals.
68 """
69 qgraph = QuantumGraph.loadUri(qgraph_uri)
70 with Butler.from_config(butler_config, writeable=False) as butler:
71 report = QuantumGraphExecutionReport.make_reports(butler, qgraph)
72 if not full_output_filename:
73 # this is the option to print to the command-line
74 summary_dict = report.to_summary_dict(butler, logs, human_readable=True)
75 dataset_table_rows = []
76 data_products = []
77 quanta_summary = []
78 error_summary = []
79 for task in summary_dict.keys():
80 for data_product in summary_dict[task]["outputs"]:
81 dataset_table_rows.append(summary_dict[task]["outputs"][data_product])
82 data_products.append(data_product)
84 if len(summary_dict[task]["failed_quanta"]) > 5:
85 quanta_summary.append(
86 {
87 "Task": task,
88 "Failed": len(summary_dict[task]["failed_quanta"]),
89 "Blocked": summary_dict[task]["n_quanta_blocked"],
90 "Succeeded": summary_dict[task]["n_succeeded"],
91 "Expected": summary_dict[task]["n_expected"],
92 }
93 )
94 else:
95 quanta_summary.append(
96 {
97 "Task": task,
98 "Failed": summary_dict[task]["failed_quanta"],
99 "Blocked": summary_dict[task]["n_quanta_blocked"],
100 "Succeeded": summary_dict[task]["n_succeeded"],
101 "Expected": summary_dict[task]["n_expected"],
102 }
103 )
104 if "errors" in summary_dict[task].keys():
105 error_summary.append({task: summary_dict[task]["errors"]})
106 quanta = Table(quanta_summary)
107 datasets = Table(dataset_table_rows)
108 datasets.add_column(data_products, index=0, name="DatasetType")
109 quanta.pprint_all()
110 print("\n")
111 if not brief:
112 pprint.pprint(error_summary)
113 print("\n")
114 datasets.pprint_all()
115 else:
116 report.write_summary_yaml(butler, full_output_filename, do_store_logs=logs)
119def report_v2(
120 butler_config: str,
121 qgraph_uris: Sequence[str],
122 collections: Sequence[str] | None,
123 where: str,
124 full_output_filename: str | None,
125 logs: bool = True,
126 brief: bool = False,
127 curse_failed_logs: bool = False,
128 read_caveats: Literal["lazy", "exhaustive"] | None = "lazy",
129 use_qbb: bool = True,
130 n_cores: int = 1,
131 view_graph: bool = False,
132) -> None:
133 """Summarize the state of executed quantum graph(s), with counts of failed,
134 successful and expected quanta, as well as counts of output datasets and
135 their visible/shadowed states. Analyze one or more attempts at the same
136 processing on the same dataquery-identified "group" and resolve recoveries
137 and persistent failures. Identify mismatch errors between groups.
139 Parameters
140 ----------
141 butler_config : `str`
142 The Butler used for this report. This should match the Butler used for
143 the run associated with the executed quantum graph.
144 qgraph_uris : `Sequence` [`str`]
145 One or more uris to the serialized Quantum Graph(s).
146 collections : `Sequence` [`str`] | None`
147 Collection(s) associated with said graphs/processing. For use in
148 `lsst.daf.butler.registry.queryDatasets` if paring down the query would
149 be useful.
150 where : `str`
151 A "where" string to use to constrain the collections, if passed.
152 full_output_filename : `str`
153 Output the full pydantic model `QuantumProvenanceGraph.Summary` object
154 into a JSON file. This is ideal for error-matching and cataloguing
155 tools such as the ones used by Campaign Management software and pilots,
156 and for searching and counting specific kinds or instances of failures.
157 This option will also print a "brief" (counts-only) summary to stdout.
158 logs : `bool`
159 Store error messages from Butler logs associated with failed quanta if
160 `True`.
161 brief : `bool`
162 Only display short (counts-only) summary on stdout. This includes
163 counts and not error messages or data_ids (similar to BPS report). This
164 option will still report all `cursed` datasets and `wonky` quanta.
165 curse_failed_logs : `bool`
166 Mark log datasets as `cursed` if they are published in the final output
167 collection. Note that a campaign-level collection must be used here for
168 `collections` if `curse_failed_logs` is `True`; if
169 `lsst.pipe.base.QuantumProvenanceGraph.__resolve_duplicates` is run on
170 a list of group-level collections, then each will only show log
171 datasets from their own failures as visible and datasets from others
172 will be marked as cursed.
173 read_caveats : `str`, optional
174 Whether and how to read success caveats from metadata datasets:
176 - "exhaustive": read all metadata datasets;
177 - "lazy": read metadata datasets only for quanta that had predicted
178 outputs that were not produced (will not pick up exceptions raised
179 after all datasets were written);
180 - `None`: do not read metadata datasets at all.
181 use_qbb : `bool`, optional
182 Whether to use a quantum-backed butler for metadata and log reads.
183 This should reduce the number of database operations.
184 n_cores : `int`, optional
185 Number of cores for metadata and log reads.
186 view_graph : `bool`
187 Display a graph representation of `QuantumProvenanceGraph.Summary` on
188 stdout instead of the default plain-text summary. Pipeline graph nodes
189 are then annotated with their status. This is a useful way to visualize
190 the flow of quanta and datasets through the graph and to identify where
191 problems may be occurring.
192 """
193 with Butler.from_config(butler_config, writeable=False) as butler:
194 qpg = QuantumProvenanceGraph(
195 butler,
196 qgraph_uris,
197 collections=collections,
198 where=where,
199 curse_failed_logs=curse_failed_logs,
200 read_caveats=read_caveats,
201 use_qbb=use_qbb,
202 n_cores=n_cores,
203 )
204 summary = qpg.to_summary(butler, do_store_logs=logs)
206 if view_graph:
207 from lsst.pipe.base.pipeline_graph.visualization import (
208 QuantumProvenanceGraphStatusAnnotator,
209 QuantumProvenanceGraphStatusOptions,
210 show,
211 )
213 # Use any of the quantum graphs to get the `PipelineGraph`
214 # representation of the pipeline.
215 qgraph = QuantumGraph.loadUri(qgraph_uris[0])
216 pipeline_graph = qgraph.pipeline_graph
218 # Annotate the pipeline graph with the status information from the
219 # quantum provenance graph summary.
220 status_annotator = QuantumProvenanceGraphStatusAnnotator(summary)
221 status_options = QuantumProvenanceGraphStatusOptions(
222 display_percent=True, display_counts=True, abbreviate=True, visualize=True
223 )
225 show(
226 pipeline_graph,
227 dataset_types=True,
228 status_annotator=status_annotator,
229 status_options=status_options,
230 )
231 else:
232 print_summary(summary, full_output_filename, brief)
235def aggregate_reports(
236 filenames: Sequence[str], full_output_filename: str | None, brief: bool = False
237) -> None:
238 """Aggregrate multiple `QuantumProvenanceGraph` summaries on separate
239 dataquery-identified groups into one wholistic report. This is intended for
240 reports over the same tasks in the same pipeline, after `pipetask report`
241 has been resolved over all graphs associated with each group.
243 Parameters
244 ----------
245 filenames : `Sequence[str]`
246 The paths to the JSON files produced by `pipetask report` (note: this
247 is only compatible with the multi-graph or `--force-v2` option). These
248 files correspond to the `QuantumProvenanceGraph.Summary` objects which
249 are produced for each group.
250 full_output_filename : `str | None`
251 The name of the JSON file in which to store the aggregate report, if
252 passed. This is passed to `print_summary` at the end of this function.
253 brief : `bool = False`
254 Only display short (counts-only) summary on stdout. This includes
255 counts and not error messages or data_ids (similar to BPS report).
256 This option will still report all `cursed` datasets and `wonky`
257 quanta. This is passed to `print_summary` at the end of this function.
258 """
259 summaries: list[Summary] = []
260 for filename in filenames:
261 with open(filename) as f:
262 model = Summary.model_validate_json(f.read())
263 summaries.extend([model])
264 result = Summary.aggregate(summaries)
265 print_summary(result, full_output_filename, brief)
268def print_summary(summary: Summary, full_output_filename: str | None, brief: bool = False) -> None:
269 """Take a `QuantumProvenanceGraph.Summary` object and write it to a file
270 and/or the screen.
272 Parameters
273 ----------
274 summary : `QuantumProvenanceGraph.Summary`
275 This `Pydantic` model contains all the information derived from the
276 `QuantumProvenanceGraph`.
277 full_output_filename : `str | None`
278 Name of the JSON file in which to store summary information, if
279 passed.
280 brief : `bool`
281 Only display short (counts-only) summary on stdout. This includes
282 counts and not error messages or data_ids (similar to BPS report).
283 Ignored (considered `False`) if ``full_output_filename`` is passed.
284 """
285 summary.pprint(brief=(brief or bool(full_output_filename)))
286 if full_output_filename:
287 with open(full_output_filename, "w") as stream:
288 stream.write(summary.model_dump_json(indent=2))