Coverage for python / lsst / pipe / base / execution_reports.py: 28%
139 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-18 01:04 -0800
1# This file is part of pipe_base.
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 program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <http://www.gnu.org/licenses/>.
21from __future__ import annotations
23__all__ = (
24 "DatasetTypeExecutionReport",
25 "QuantumGraphExecutionReport",
26 "TaskExecutionReport",
27 "lookup_quantum_data_id",
28)
30import dataclasses
31import itertools
32import logging
33import uuid
34from collections.abc import Iterable, Mapping
35from typing import Any
37import networkx
38import yaml
40from lsst.daf.butler import Butler, DataCoordinate, DatasetRef, Quantum
41from lsst.resources import ResourcePathExpression
43from .graph import QuantumGraph
46@dataclasses.dataclass
47class DatasetTypeExecutionReport:
48 """A report on the number of produced datasets as well as the status of
49 missing datasets based on metadata.
51 A `DatasetTypeExecutionReport` is created for each
52 `~lsst.daf.butler.DatasetType` in a `TaskExecutionReport`.
53 """
55 failed: set[DatasetRef] = dataclasses.field(default_factory=set)
56 """Datasets not produced because their quanta failed directly in this
57 run (`set`).
58 """
60 not_produced: set[DatasetRef] = dataclasses.field(default_factory=set)
61 """Missing datasets which were not produced by successful quanta.
62 """
64 blocked: set[DatasetRef] = dataclasses.field(default_factory=set)
65 """Datasets not produced due to an upstream failure (`set`).
66 """
68 n_produced: int = 0
69 """Count of datasets produced (`int`).
70 """
72 n_expected: int = 0
73 """Count of datasets expected (`int`)
74 """
76 def to_summary_dict(self) -> dict[str, Any]:
77 r"""Summarize the DatasetTypeExecutionReport in a dictionary.
79 Returns
80 -------
81 summary_dict : `dict`
82 A count of the datasets with each outcome; the number of
83 produced, ``failed``, ``not_produced``, and ``blocked``
84 `~lsst.daf.butler.DatasetType`\ s.
85 See above for attribute descriptions.
86 """
87 return {
88 "produced": self.n_produced,
89 "failed": len(self.failed),
90 "not_produced": len(self.not_produced),
91 "blocked": len(self.blocked),
92 "expected": self.n_expected,
93 }
96@dataclasses.dataclass
97class TaskExecutionReport:
98 """A report on the status and content of a task in an executed quantum
99 graph.
101 Use task metadata to identify and inspect failures and report on output
102 datasets.
104 See Also
105 --------
106 QuantumGraphExecutionReport : Quantum graph report.
107 DatasetTypeExecutionReport : DatasetType report.
108 """
110 failed: dict[uuid.UUID, DatasetRef] = dataclasses.field(default_factory=dict)
111 """A mapping from quantum data ID to log dataset reference for quanta that
112 failed directly in this run (`dict`).
113 """
115 n_succeeded: int = 0
116 """A count of successful quanta.
118 This may include quanta that did not produce any datasets; ie, raised
119 `NoWorkFound`.
120 """
122 n_expected: int = 0
123 """A count of expected quanta.
124 """
126 blocked: dict[uuid.UUID, DataCoordinate] = dataclasses.field(default_factory=dict)
127 """A mapping of data IDs of quanta that were not attempted due to an
128 upstream failure (`dict`).
129 """
131 output_datasets: dict[str, DatasetTypeExecutionReport] = dataclasses.field(default_factory=dict)
132 """Missing and produced outputs of each `~lsst.daf.butler.DatasetType`
133 (`dict`).
134 """
136 def inspect_quantum(
137 self,
138 quantum_id: uuid.UUID,
139 quantum: Quantum,
140 status_graph: networkx.DiGraph,
141 refs: Mapping[str, Mapping[uuid.UUID, DatasetRef]],
142 metadata_name: str,
143 log_name: str,
144 ) -> None:
145 """Inspect a quantum of a quantum graph and ascertain the status of
146 each associated data product.
148 Parameters
149 ----------
150 quantum_id : `uuid.UUID`
151 Unique identifier for the quantum to inspect.
152 quantum : `Quantum`
153 The specific node of the quantum graph to be inspected.
154 status_graph : `networkx.DiGraph`
155 The quantum graph produced by
156 `QuantumGraphExecutionReport.make_reports` which steps through the
157 quantum graph of a run and logs the status of each quantum.
158 refs : `~collections.abc.Mapping` [ `str`,\
159 `~collections.abc.Mapping` [ `uuid.UUID`,\
160 `~lsst.daf.butler.DatasetRef` ] ]
161 The DatasetRefs of each of the DatasetTypes produced by the task.
162 Includes initialization, intermediate and output data products.
163 metadata_name : `str`
164 The metadata dataset name for the node.
165 log_name : `str`
166 The name of the log files for the node.
168 See Also
169 --------
170 QuantumGraphExecutionReport.make_reports : Make reports.
171 """
172 (metadata_ref,) = quantum.outputs[metadata_name]
173 (log_ref,) = quantum.outputs[log_name]
174 blocked = False
175 if metadata_ref.id not in refs[metadata_name]:
176 if any(
177 status_graph.nodes[upstream_quantum_id]["failed"]
178 for upstream_dataset_id in status_graph.predecessors(quantum_id)
179 for upstream_quantum_id in status_graph.predecessors(upstream_dataset_id)
180 ):
181 assert quantum.dataId is not None
182 self.blocked[quantum_id] = quantum.dataId
183 blocked = True
184 else:
185 self.failed[quantum_id] = log_ref
186 # note: log_ref may or may not actually exist
187 failed = True
188 else:
189 failed = False
190 self.n_succeeded += 1
191 status_graph.nodes[quantum_id]["failed"] = failed
193 # Now, loop over the datasets to make a DatasetTypeExecutionReport.
194 for output_ref in itertools.chain.from_iterable(quantum.outputs.values()):
195 if output_ref == metadata_ref or output_ref == log_ref:
196 continue
197 if (dataset_type_report := self.output_datasets.get(output_ref.datasetType.name)) is None:
198 dataset_type_report = DatasetTypeExecutionReport()
199 self.output_datasets[output_ref.datasetType.name] = dataset_type_report
200 if output_ref.id not in refs[output_ref.datasetType.name]:
201 if failed:
202 if blocked:
203 dataset_type_report.blocked.add(output_ref)
204 else:
205 dataset_type_report.failed.add(output_ref)
206 else:
207 dataset_type_report.not_produced.add(output_ref)
208 else:
209 dataset_type_report.n_produced += 1
210 dataset_type_report.n_expected += 1
212 def to_summary_dict(
213 self, butler: Butler, do_store_logs: bool = True, human_readable: bool = False
214 ) -> dict[str, Any]:
215 """Summarize the results of the TaskExecutionReport in a dictionary.
217 Parameters
218 ----------
219 butler : `lsst.daf.butler.Butler`
220 The Butler used for this report.
221 do_store_logs : `bool`
222 Store the logs in the summary dictionary.
223 human_readable : `bool`
224 Store more human-readable information to be printed out to the
225 command-line.
227 Returns
228 -------
229 summary_dict : `dict`
230 A dictionary containing:
232 - outputs: A dictionary summarizing the
233 DatasetTypeExecutionReport for each DatasetType associated with
234 the task
235 - failed_quanta: A dictionary of quanta which failed and their
236 dataIDs by quantum graph node id
237 - n_quanta_blocked: The number of quanta which failed due to
238 upstream failures.
239 - n_succeded: The number of quanta which succeeded.
241 And possibly, if human-readable is passed:
243 - errors: A dictionary of data ids associated with each error
244 message. If `human-readable` and `do_store_logs`, this is stored
245 here. Otherwise, if `do_store_logs`, it is stored in
246 `failed_quanta` keyed by the quantum graph node id.
247 """
248 failed_quanta = {}
249 failed_data_ids = []
250 errors = []
251 for node_id, log_ref in self.failed.items():
252 data_id = dict(log_ref.dataId.required)
253 quantum_info: dict[str, Any] = {"data_id": data_id}
254 if do_store_logs:
255 try:
256 log = butler.get(log_ref)
257 except LookupError:
258 quantum_info["error"] = []
259 except FileNotFoundError:
260 quantum_info["error"] = None
261 else:
262 quantum_info["error"] = [
263 record.message for record in log if record.levelno >= logging.ERROR
264 ]
265 if human_readable:
266 failed_data_ids.append(data_id)
267 if do_store_logs:
268 errors.append(quantum_info)
270 else:
271 failed_quanta[str(node_id)] = quantum_info
272 result = {
273 "outputs": {name: r.to_summary_dict() for name, r in self.output_datasets.items()},
274 "n_quanta_blocked": len(self.blocked),
275 "n_succeeded": self.n_succeeded,
276 "n_expected": self.n_expected,
277 }
278 if human_readable:
279 result["failed_quanta"] = failed_data_ids
280 result["errors"] = errors
281 else:
282 result["failed_quanta"] = failed_quanta
283 return result
285 def __str__(self) -> str:
286 """Return a count of the failed and blocked tasks in the
287 TaskExecutionReport.
288 """
289 return f"failed: {len(self.failed)}\nblocked: {len(self.blocked)}\n"
292@dataclasses.dataclass
293class QuantumGraphExecutionReport:
294 """A report on the execution of a quantum graph.
296 Report the detailed status of each failure; whether tasks were not run,
297 data is missing from upstream failures, or specific errors occurred during
298 task execution (and report the errors). Contains a count of expected,
299 produced DatasetTypes for each task. This report can be output as a
300 dictionary or a yaml file.
302 Attributes
303 ----------
304 tasks : `dict`
305 A dictionary of TaskExecutionReports by task label.
307 See Also
308 --------
309 TaskExecutionReport : A task report.
310 DatasetTypeExecutionReport : A dataset type report.
311 """
313 tasks: dict[str, TaskExecutionReport] = dataclasses.field(default_factory=dict)
314 """A dictionary of TaskExecutionReports by task label (`dict`)."""
316 def to_summary_dict(
317 self, butler: Butler, do_store_logs: bool = True, human_readable: bool = False
318 ) -> dict[str, Any]:
319 """Summarize the results of the `QuantumGraphExecutionReport` in a
320 dictionary.
322 Parameters
323 ----------
324 butler : `lsst.daf.butler.Butler`
325 The Butler used for this report.
326 do_store_logs : `bool`
327 Store the logs in the summary dictionary.
328 human_readable : `bool`
329 Store more human-readable information to be printed out to the
330 command-line.
332 Returns
333 -------
334 summary_dict : `dict`
335 A dictionary containing a summary of a `TaskExecutionReport` for
336 each task in the quantum graph.
337 """
338 return {
339 task: report.to_summary_dict(butler, do_store_logs=do_store_logs, human_readable=human_readable)
340 for task, report in self.tasks.items()
341 }
343 def write_summary_yaml(self, butler: Butler, filename: str, do_store_logs: bool = True) -> None:
344 """Take the dictionary from
345 `QuantumGraphExecutionReport.to_summary_dict` and store its contents in
346 a yaml file.
348 Parameters
349 ----------
350 butler : `lsst.daf.butler.Butler`
351 The Butler used for this report.
352 filename : `str`
353 The name to be used for the summary yaml file.
354 do_store_logs : `bool`
355 Store the logs in the summary dictionary.
356 """
357 with open(filename, "w") as stream:
358 yaml.safe_dump(self.to_summary_dict(butler, do_store_logs=do_store_logs), stream)
360 @classmethod
361 def make_reports(
362 cls,
363 butler: Butler,
364 graph: QuantumGraph | ResourcePathExpression,
365 ) -> QuantumGraphExecutionReport:
366 """Make a `QuantumGraphExecutionReport`.
368 Step through the quantum graph associated with a run, creating a
369 `networkx.DiGraph` called status_graph to annotate the status of each
370 quantum node. For each task in the quantum graph, use
371 `TaskExecutionReport.inspect_quantum` to make a `TaskExecutionReport`
372 based on the status of each node. Return a `TaskExecutionReport` for
373 each task in the quantum graph.
375 Parameters
376 ----------
377 butler : `lsst.daf.butler.Butler`
378 The Butler used for this report. This should match the Butler used
379 for the run associated with the executed quantum graph.
380 graph : `QuantumGraph` | `ResourcePathExpression`
381 Either the associated quantum graph object or the uri of the
382 location of said quantum graph.
384 Returns
385 -------
386 report: `QuantumGraphExecutionReport`
387 The `TaskExecutionReport` for each task in the quantum graph.
388 """
389 refs = {} # type: dict[str, Any]
390 status_graph = networkx.DiGraph()
391 if not isinstance(graph, QuantumGraph):
392 qg = QuantumGraph.loadUri(graph)
393 else:
394 qg = graph
395 assert qg.metadata is not None, "Saved QGs always have metadata."
396 collection = qg.metadata["output_run"]
397 report = cls()
398 for dataset_type_node in qg.pipeline_graph.dataset_types.values():
399 if qg.pipeline_graph.producer_of(dataset_type_node.name) is None:
400 continue
401 refs[dataset_type_node.name] = {
402 ref.id: ref
403 for ref in butler.registry.queryDatasets(
404 dataset_type_node.name, collections=collection, findFirst=False
405 )
406 }
407 for task_node in qg.pipeline_graph.tasks.values():
408 for quantum_id, quantum in qg.get_task_quanta(task_node.label).items():
409 status_graph.add_node(quantum_id)
410 for ref in itertools.chain.from_iterable(quantum.outputs.values()):
411 status_graph.add_edge(quantum_id, ref.id)
412 for ref in itertools.chain.from_iterable(quantum.inputs.values()):
413 status_graph.add_edge(ref.id, quantum_id)
414 for task_node in qg.pipeline_graph.tasks.values():
415 task_report = TaskExecutionReport()
416 if task_node.log_output is None:
417 raise RuntimeError("QG must have log outputs to use execution reports.")
418 for quantum_id, quantum in qg.get_task_quanta(task_node.label).items():
419 task_report.inspect_quantum(
420 quantum_id,
421 quantum,
422 status_graph,
423 refs,
424 metadata_name=task_node.metadata_output.dataset_type_name,
425 log_name=task_node.log_output.dataset_type_name,
426 )
427 task_report.n_expected = len(qg.get_task_quanta(task_node.label).items())
428 report.tasks[task_node.label] = task_report
429 return report
431 def __str__(self) -> str:
432 return "\n".join(f"{tasklabel}:{report}" for tasklabel, report in self.tasks.items())
435def lookup_quantum_data_id(
436 graph_uri: ResourcePathExpression, nodes: Iterable[uuid.UUID]
437) -> list[DataCoordinate | None]:
438 """Look up a dataId from a quantum graph and a list of quantum graph
439 nodeIDs.
441 Parameters
442 ----------
443 graph_uri : `ResourcePathExpression`
444 URI of the quantum graph of the run.
445 nodes : `~collections.abc.Iterable` [ `uuid.UUID` ]
446 Quantum graph nodeID.
448 Returns
449 -------
450 data_ids : `list` [ `lsst.daf.butler.DataCoordinate` ]
451 A list of human-readable dataIDs which map to the nodeIDs on the
452 quantum graph at graph_uri.
453 """
454 qg = QuantumGraph.loadUri(graph_uri, nodes=nodes)
455 return [qg.getQuantumNodeByNodeId(node).quantum.dataId for node in nodes]