Coverage for python / lsst / pipe / base / quantum_provenance_graph.py: 26%
720 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-09 01:22 -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 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/>.
28"""A set of already-run, merged quantum graphs with provenance information
29which can be used to compose a report on the status of multi-attempt
30processing.
31"""
33from __future__ import annotations
35__all__ = (
36 "DatasetKey",
37 "PrerequisiteDatasetKey",
38 "QuantumKey",
39 "QuantumProvenanceGraph",
40)
42import concurrent.futures
43import dataclasses
44import datetime
45import itertools
46import logging
47import textwrap
48import threading
49import uuid
50from collections.abc import Callable, Iterator, Mapping, Sequence, Set
51from enum import Enum
52from typing import Any, ClassVar, Literal, TypedDict, cast
54import astropy.table
55import networkx
56import pydantic
58from lsst.daf.butler import (
59 Butler,
60 ButlerConfig,
61 ButlerLogRecords,
62 DataCoordinate,
63 DataIdValue,
64 DatasetId,
65 DatasetRef,
66 DatasetType,
67 DimensionUniverse,
68 LimitedButler,
69 MissingDatasetTypeError,
70 QuantumBackedButler,
71)
72from lsst.resources import ResourcePathExpression
73from lsst.utils.logging import PeriodicLogger, getLogger
75from ._status import ExceptionInfo, QuantumSuccessCaveats
76from .automatic_connection_constants import (
77 LOG_OUTPUT_CONNECTION_NAME,
78 LOG_OUTPUT_TEMPLATE,
79 METADATA_OUTPUT_CONNECTION_NAME,
80 METADATA_OUTPUT_STORAGE_CLASS,
81 METADATA_OUTPUT_TEMPLATE,
82)
83from .graph import QuantumGraph, QuantumNode
85_LOG = getLogger(__name__)
88@dataclasses.dataclass(slots=True, eq=True, frozen=True)
89class QuantumKey:
90 """Identifier type for quantum keys in a `QuantumProvenanceGraph`. These
91 keys correspond to a task label and data ID, but can refer to this over
92 multiple runs or datasets.
93 """
95 task_label: str
96 """Label of the task in the pipeline."""
98 data_id_values: tuple[DataIdValue, ...]
99 """Data ID values of the quantum.
101 Note that keys are fixed given `task_label`, so using only the values here
102 speeds up comparisons.
103 """
105 is_task: ClassVar[Literal[True]] = True
106 """Whether this node represents a quantum rather
107 than a dataset (always `True`).
108 """
111@dataclasses.dataclass(slots=True, eq=True, frozen=True)
112class DatasetKey:
113 """Identifier type for dataset keys in a `QuantumProvenanceGraph`."""
115 dataset_type_name: str
116 """Name of the dataset type (never a component)."""
118 data_id_values: tuple[DataIdValue, ...]
119 """Data ID values of the dataset.
121 Note that keys are fixed given `parent_dataset_type_name`, so using only
122 the values here speeds up comparisons.
123 """
125 is_task: ClassVar[Literal[False]] = False
126 """Whether this node represents a quantum rather than a dataset (always
127 `False`).
128 """
130 is_prerequisite: ClassVar[Literal[False]] = False
131 """Whether this node is a prerequisite to another node (also always
132 `False`).
133 """
136@dataclasses.dataclass(slots=True, eq=True, frozen=True)
137class PrerequisiteDatasetKey:
138 """Identifier type for prerequisite dataset keys in a
139 `QuantumProvenanceGraph`.
141 Unlike regular datasets, prerequisites are not actually required to come
142 from a find-first search of `input_collections`, so we don't want to
143 assume that the same data ID implies the same dataset. Happily we also
144 don't need to search for them by data ID in the graph, so we can use the
145 dataset ID (UUID) instead.
146 """
148 dataset_type_name: str
149 """Name of the dataset type (never a component)."""
151 dataset_id_bytes: bytes
152 """Dataset ID (UUID) as raw bytes."""
154 is_task: ClassVar[Literal[False]] = False
155 """Whether this node represents a quantum rather
156 than a dataset (always `False`).
157 """
159 is_prerequisite: ClassVar[Literal[True]] = True
160 """Whether this node is a prerequisite to another node (always `True`).
161 """
164class QuantumRunStatus(Enum):
165 """Enum describing the status of a quantum-run collection combination.
167 Possible Statuses
168 -----------------
169 METADATA_MISSING = -3: Metadata is missing for this quantum in this run.
170 It is impossible to tell whether execution of this quantum was
171 attempted due to missing metadata.
172 LOGS_MISSING = -2: Logs are missing for this quantum in this run. It was
173 attempted, but it is impossible to tell if it succeeded or failed due
174 to missing logs.
175 FAILED = -1: Attempts to execute the quantum failed in this run.
176 BLOCKED = 0: This run does not include an executed version of this
177 quantum because an upstream task failed.
178 SUCCESSFUL = 1: This quantum was executed successfully in this run.
179 """
181 METADATA_MISSING = -3
182 LOGS_MISSING = -2
183 FAILED = -1
184 BLOCKED = 0
185 SUCCESSFUL = 1
188class QuantumRun(pydantic.BaseModel):
189 """Information about a quantum in a given run collection."""
191 model_config = pydantic.ConfigDict(arbitrary_types_allowed=True) # for DatasetRef attrs.
193 id: uuid.UUID
194 """The quantum graph node ID associated with the dataId in a specific run.
195 """
197 status: QuantumRunStatus = QuantumRunStatus.METADATA_MISSING
198 """The status of the quantum in that run.
199 """
201 caveats: QuantumSuccessCaveats | None = None
202 """Flags that describe possibly-qualified successes.
204 This is `None` when `status` is not `SUCCESSFUL` or `LOGS_MISSING`. It
205 may also be `None` if metadata was not loaded or had no success flags.
206 """
208 exception: ExceptionInfo | None = None
209 """Information about an exception that that was raised during the quantum's
210 execution.
212 Exception information for failed quanta is not currently stored, so this
213 field is actually only populated for quanta that raise
214 `AnnotatedPartialOutputsError`, and only when the execution system is
215 configured not to consider these partial successes a failure (i.e. when
216 `status` is `~QuantumRunStatus.SUCCESSFUL` and `caveats` has
217 `~QuantumSuccessCaveats.PARTIAL_OUTPUTS_ERROR` set. The error whose
218 information is reported here is the exception chained from the
219 `AnnotatedPartialOutputsError`.
221 In the future, exception information from failures may be available as
222 well.
223 """
225 metadata_ref: DatasetRef
226 """Predicted DatasetRef for the metadata dataset."""
228 log_ref: DatasetRef
229 """Predicted DatasetRef for the log dataset."""
231 @staticmethod
232 def find_final(info: QuantumInfo) -> tuple[str, QuantumRun]:
233 """Return the final RUN collection name and `QuantumRun` structure from
234 a `QuantumInfo` dictionary.
236 The "final run" is the last RUN collection in the sequence of quantum
237 graphs that:
239 - actually had a quantum for that task label and data ID;
240 - execution seems to have at least been attempted (at least one of
241 metadata or logs were produced).
243 Parameters
244 ----------
245 info : `QuantumInfo`
246 Quantum information that includes all runs.
248 Returns
249 -------
250 run : `str`
251 RUN collection name.
252 quantum_run : `QuantumRun`
253 Information about a quantum in a RUN collection.
255 Raises
256 ------
257 ValueError
258 Raised if this quantum never had a status that suggested execution
259 in any run.
260 """
261 for run, quantum_run in reversed(info["runs"].items()):
262 if (
263 quantum_run.status is not QuantumRunStatus.METADATA_MISSING
264 and quantum_run.status is not QuantumRunStatus.BLOCKED
265 ):
266 return run, quantum_run
267 raise ValueError("Quantum was never executed.")
270class QuantumInfoStatus(Enum):
271 """The status of a quantum (a particular task run on a particular dataID)
272 across all runs.
274 Possible Statuses
275 -----------------
276 WONKY = -3: The overall state of this quantum reflects inconsistencies or
277 is difficult to discern. There are a few specific ways to enter a wonky
278 state; it is impossible to exit and requires human intervention to
279 proceed with processing.
280 Currently, a quantum enters a wonky state for one of three reasons:
281 - Its overall `QuantumInfoStatus` moves from a successful state (as a
282 result of a successful run) to any other state. In other words,
283 something that initially succeeded fails on subsequent attempts.
284 - A `QuantumRun` is missing logs.
285 - There are multiple runs associated with a dataset, and this comes up
286 in a findFirst search. This means that a dataset which will be used
287 as an input data product for further processing has heterogeneous
288 inputs, which may have had different inputs or a different
289 data-query.
290 FAILED = -2: These quanta were attempted and failed. Failed quanta have
291 logs and no metadata.
292 UNKNOWN = -1: These are quanta which do not have any metadata associated
293 with processing, but for which it is impossible to tell the status due
294 to an additional absence of logs. Quanta which had not been processed
295 at all would reflect this state, as would quanta which were
296 conceptualized in the construction of the quantum graph but later
297 identified to be unneccesary or erroneous (deemed NoWorkFound by the
298 Science Pipelines).
299 BLOCKED = 0: The quantum is not able to execute because its inputs are
300 missing due to an upstream failure. Blocked quanta are distinguished
301 from failed quanta by being successors of failed quanta in the graph.
302 All the successors of blocked quanta are also marked as blocked.
303 SUCCESSFUL = 1: Attempts at executing this quantum were successful.
304 """
306 WONKY = -3
307 FAILED = -2
308 UNKNOWN = -1
309 BLOCKED = 0
310 SUCCESSFUL = 1
313class QuantumInfo(TypedDict):
314 """Information about a quantum (i.e., the combination of a task label and
315 data ID) across all attempted runs.
317 Used to annotate the networkx node dictionary.
318 """
320 data_id: DataCoordinate
321 """The data_id of the quantum.
322 """
324 runs: dict[str, QuantumRun]
325 """All run collections associated with the quantum.
326 """
328 status: QuantumInfoStatus
329 """The overall status of the quantum. Note that it is impossible to exit a
330 wonky state.
331 """
333 recovered: bool
334 """The quantum was originally not successful but was ultimately successful.
335 """
337 messages: list[str]
338 """Diagnostic messages to help disambiguate wonky states.
339 """
341 log: DatasetKey
342 """The `DatasetKey` which can be used to access the log associated with the
343 quantum across runs.
344 """
346 metadata: DatasetKey
347 """The `DatasetKey` which can be used to access the metadata for the
348 quantum across runs.
349 """
352class DatasetRun(pydantic.BaseModel):
353 """Information about a dataset in a given run collection."""
355 id: uuid.UUID
356 """The dataset ID associated with the dataset in a specific run.
357 """
359 produced: bool = False
360 """Whether the specific run wrote the dataset.
361 """
363 visible: bool = False
364 """Whether this dataset is visible in the final output collection; in other
365 words, whether this dataset is queryable in a find-first search. This
366 determines whether it will be used as an input to further processing.
367 """
369 @pydantic.model_validator(mode="after")
370 def _validate(self) -> DatasetRun:
371 """Validate the model for `DatasetRun` by asserting that no visible
372 `DatasetRun` is also not produced (this should be impossible).
374 Returns
375 -------
376 self : `DatasetRun`
377 The `DatasetRun` object, validated.
378 """
379 assert not (self.visible and not self.produced)
380 return self
383class DatasetInfoStatus(Enum):
384 """Status of the the DatasetType-dataID pair over all runs. This depends
385 not only on the presence of the dataset itself, but also on metadata, logs
386 and the state of its producer quantum.
388 Possible Statuses
389 -----------------
390 CURSED: The dataset was the result of an unsuccessful quantum and was
391 visible in the output collection anyway. These are flagged as
392 cursed so that they may be caught before they become inputs to
393 further processing.
394 UNSUCCESSFUL: The dataset was not produced. These are the results of
395 failed or blocked quanta.
396 PREDICTED_ONLY: The dataset was predicted, and was not visible in any
397 run, but was the successor of a successful quantum. These datasets are
398 the result of pipelines NoWorkFound cases, in which a dataset is
399 predicted in the graph but found to not be necessary in processing.
400 SHADOWED: The dataset exists but is not queryable in a find_first
401 search. This could mean that the version of this dataset which is
402 passed as an input to further processing is not in the collections
403 given. A shadowed dataset will not be used as an input to further
404 processing.
405 VISIBLE: The dataset is queryable in a find_first search. This means
406 that it can be used as an input by subsequent tasks and processing.
407 """
409 CURSED = -2
410 UNSUCCESSFUL = -1
411 PREDICTED_ONLY = 0
412 SHADOWED = 1
413 VISIBLE = 2
416class DatasetInfo(TypedDict):
417 """Information about a given dataset across all runs.
419 Used to annotate the networkx node dictionary.
420 """
422 data_id: DataCoordinate
423 """The data_id of the quantum.
424 """
426 runs: dict[str, DatasetRun]
427 """All runs associated with the dataset.
428 """
430 status: DatasetInfoStatus
431 """Overall status of the dataset.
432 """
434 messages: list[str]
435 """Diagnostic messages to help disambiguate cursed states.
436 """
439class UnsuccessfulQuantumSummary(pydantic.BaseModel):
440 """A summary of all relevant information on an unsuccessful quantum.
442 This summarizes all information on a task's output for a particular data ID
443 over all runs.
444 """
446 data_id: dict[str, DataIdValue]
447 """The data_id of the unsuccessful quantum.
448 """
449 runs: dict[str, str]
450 """A dictionary (key: output run collection name) with the value of the
451 enum name of the `QuantumRunStatus` of each run associated with an attempt
452 to process the unsuccessful quantum.
453 """
454 messages: list[str]
455 """Any messages associated with the unsuccessful quantum (any clues as to
456 why the quantum may be in a FAILED or WONKY state).
457 """
459 @classmethod
460 def _from_info(cls, info: QuantumInfo) -> UnsuccessfulQuantumSummary:
461 """Summarize all relevant information from the `QuantumInfo` in an
462 `UnsuccessfulQuantumSummary`; return an `UnsuccessfulQuantumSummary`.
464 Parameters
465 ----------
466 info : `QuantumInfo`
467 The `QuantumInfo` object for the unsuccessful quantum.
469 Returns
470 -------
471 summary : `UnsuccessfulQuantumSummary`
472 A Pydantic model containing the dataID, run collection names (and
473 each of their `QuantumRunStatus` enum names) as well as messages
474 which may point to any clues about the nature of the problem. For
475 failed quanta, these are usually error messages from the butler
476 logs. For wonky quanta, these can be messages generated during the
477 assembly of the `QuantumProvenanceGraph` that describe why it was
478 marked as wonky.
479 """
480 return cls(
481 data_id=dict(info["data_id"].required),
482 runs={k: v.status.name for k, v in info["runs"].items()},
483 messages=info["messages"],
484 )
487class ExceptionInfoSummary(pydantic.BaseModel):
488 """A summary of an exception raised by a quantum."""
490 quantum_id: uuid.UUID
491 """Unique identifier for this quantum in this run."""
493 data_id: dict[str, DataIdValue]
494 """The data ID of the quantum."""
496 run: str
497 """Name of the RUN collection in which this exception was (last) raised."""
499 exception: ExceptionInfo
500 """Information about an exception chained to
501 `AnnotatedPartialOutputsError`, if that was raised by this quantum in this
502 run.
503 """
506class TaskSummary(pydantic.BaseModel):
507 """A summary of the status of all quanta associated with a single task,
508 across all runs.
509 """
511 n_successful: int = 0
512 """A count of successful quanta.
513 """
514 n_blocked: int = 0
515 """A count of blocked quanta.
516 """
517 n_unknown: int = 0
518 """A count of quanta for which there are no metadata or logs.
519 """
521 n_expected: int = 0
522 """The number of quanta expected by the graph.
523 """
525 @pydantic.computed_field # type: ignore[prop-decorator]
526 @property
527 def n_wonky(self) -> int:
528 """Return a count of `wonky` quanta."""
529 return len(self.wonky_quanta)
531 @pydantic.computed_field # type: ignore[prop-decorator]
532 @property
533 def n_failed(self) -> int:
534 """Return a count of `failed` quanta."""
535 return len(self.failed_quanta)
537 caveats: dict[str, list[dict[str, DataIdValue]]] = pydantic.Field(default_factory=dict)
538 """Quanta that were successful with caveats.
540 Keys are 2-character codes returned by `QuantumSuccessCaveats.concise`;
541 values are lists of data IDs of quanta with those caveats. Quanta that were
542 unqualified successes are not included.
544 Quanta for which success flags were not read from metadata will not be
545 included.
546 """
548 exceptions: dict[str, list[ExceptionInfoSummary]] = pydantic.Field(default_factory=dict)
549 """Exceptions raised by partially-successful quanta.
551 Keys are fully-qualified exception type names and values are lists of
552 extra exception information, with each entry corresponding to different
553 data ID. Only the final RUN for each data ID is represented here.
555 Every entry in this data structure corresponds to one in `ceveats` with
556 the "P" code (for `QuantumSuccessCaveats.PARTIAL_OUTPUTS_ERROR`).
558 In the future, this may be expanded to include exceptions for failed quanta
559 as well (at present that information is not retained during execution).
560 """
562 failed_quanta: list[UnsuccessfulQuantumSummary] = pydantic.Field(default_factory=list)
563 """A list of all `UnsuccessfulQuantumSummary` objects associated with the
564 FAILED quanta. This is a report containing their data IDs, the status
565 of each run associated with each `failed` quantum, and the error messages
566 associated with the failures when applicable.
567 """
568 recovered_quanta: list[dict[str, DataIdValue]] = pydantic.Field(default_factory=list)
569 """A list of dataIDs (key->value) which moved from an unsuccessful to
570 successful state.
571 """
572 wonky_quanta: list[UnsuccessfulQuantumSummary] = pydantic.Field(default_factory=list)
573 """A list of all `UnsuccessfulQuantumSummary` objects associated with the
574 WONKY quanta. This is a report containing their data_ids, the status of
575 each run associated with each `wonky` quantum, and messages (dictated in
576 this module) associated with the particular issue identified.
577 """
579 def _add_quantum_info(
580 self,
581 info: QuantumInfo,
582 log_getter: Callable[[DatasetRef], ButlerLogRecords] | None,
583 executor: concurrent.futures.Executor,
584 ) -> concurrent.futures.Future[None] | None:
585 """Add a `QuantumInfo` to a `TaskSummary`.
587 Unpack the `QuantumInfo` object, sorting quanta of each status into
588 the correct place in the `TaskSummary`. If looking for error messages
589 in the `Butler` logs is desired, take special care to catch issues
590 with missing logs.
592 Parameters
593 ----------
594 info : `QuantumInfo`
595 The `QuantumInfo` object to add to the `TaskSummary`.
596 log_getter : `~collections.abc.Callable` or `None`
597 A callable that can be passed a `~lsst.daf.butler.DatasetRef` for
598 a log dataset to retreive those logs, or `None` to not load any
599 logs.
600 executor : `concurrent.futures.Executor`
601 A possibly-parallel executor that should be used to schedule
602 log dataset reads.
604 Returns
605 -------
606 future : `concurrent.futures.Future` or `None`
607 A future that represents a parallelized log read and summary
608 update.
609 """
610 try:
611 final_run, final_quantum_run = QuantumRun.find_final(info)
612 except ValueError:
613 final_run = None
614 final_quantum_run = None
615 match info["status"]:
616 case QuantumInfoStatus.SUCCESSFUL:
617 self.n_successful += 1
618 if info["recovered"]:
619 self.recovered_quanta.append(dict(info["data_id"].required))
620 if final_quantum_run is not None and final_quantum_run.caveats:
621 code = final_quantum_run.caveats.concise()
622 self.caveats.setdefault(code, []).append(dict(info["data_id"].required))
623 if final_quantum_run.caveats & QuantumSuccessCaveats.PARTIAL_OUTPUTS_ERROR:
624 if final_quantum_run.exception is not None:
625 self.exceptions.setdefault(final_quantum_run.exception.type_name, []).append(
626 ExceptionInfoSummary(
627 quantum_id=final_quantum_run.id,
628 data_id=dict(info["data_id"].required),
629 run=final_run,
630 exception=final_quantum_run.exception,
631 )
632 )
633 return None
634 case QuantumInfoStatus.WONKY:
635 self.wonky_quanta.append(UnsuccessfulQuantumSummary._from_info(info))
636 return None
637 case QuantumInfoStatus.BLOCKED:
638 self.n_blocked += 1
639 return None
640 case QuantumInfoStatus.FAILED:
641 failed_quantum_summary = UnsuccessfulQuantumSummary._from_info(info)
642 future: concurrent.futures.Future[None] | None = None
643 if log_getter:
645 def callback() -> None:
646 for quantum_run in info["runs"].values():
647 try:
648 log = log_getter(quantum_run.log_ref)
649 except LookupError:
650 failed_quantum_summary.messages.append(
651 f"Logs not ingested for {quantum_run.log_ref!r}"
652 )
653 except FileNotFoundError:
654 failed_quantum_summary.messages.append(
655 f"Logs missing or corrupt for {quantum_run.log_ref!r}"
656 )
657 else:
658 failed_quantum_summary.messages.extend(
659 [record.message for record in log if record.levelno >= logging.ERROR]
660 )
662 future = executor.submit(callback)
663 self.failed_quanta.append(failed_quantum_summary)
664 return future
665 case QuantumInfoStatus.UNKNOWN:
666 self.n_unknown += 1
667 return None
668 case unrecognized_state:
669 raise AssertionError(f"Unrecognized quantum status {unrecognized_state!r}")
671 def _add_data_id_group(self, other_summary: TaskSummary) -> None:
672 """Add information from a `TaskSummary` over one dataquery-identified
673 group to another, as part of aggregating `Summary` reports.
675 Parameters
676 ----------
677 other_summary : `TaskSummary`
678 `TaskSummary` to aggregate.
679 """
680 self.n_successful += other_summary.n_successful
681 self.n_blocked += other_summary.n_blocked
682 self.n_unknown += other_summary.n_unknown
683 self.n_expected += other_summary.n_expected
684 for code in self.caveats.keys() | other_summary.caveats.keys():
685 self.caveats.setdefault(code, []).extend(other_summary.caveats.get(code, []))
686 for type_name in self.exceptions.keys() | other_summary.exceptions.keys():
687 self.exceptions.setdefault(type_name, []).extend(other_summary.exceptions.get(type_name, []))
688 self.wonky_quanta.extend(other_summary.wonky_quanta)
689 self.recovered_quanta.extend(other_summary.recovered_quanta)
690 self.failed_quanta.extend(other_summary.failed_quanta)
693class CursedDatasetSummary(pydantic.BaseModel):
694 """A summary of all the relevant information on a cursed dataset."""
696 producer_data_id: dict[str, DataIdValue]
697 """The data_id of the task which produced this dataset. This is mostly
698 useful for people wishing to track down the task which produced this
699 cursed dataset quickly.
700 """
701 data_id: dict[str, DataIdValue]
702 """The data_id of the cursed dataset.
703 """
704 runs_produced: dict[str, bool]
705 """A dictionary of all the runs associated with the cursed dataset;
706 the `bool` is true if the dataset was produced in the associated run.
707 """
708 run_visible: str | None
709 """The run collection that holds the dataset that is visible in the final
710 output collection.
711 """
712 messages: list[str]
713 """Any diagnostic messages (dictated in this module) which might help in
714 understanding why or how the dataset became cursed.
715 """
717 @classmethod
718 def _from_info(cls, info: DatasetInfo, producer_info: QuantumInfo) -> CursedDatasetSummary:
719 """Summarize all relevant information from the `DatasetInfo` in an
720 `CursedDatasetSummary`; return a `CursedDatasetSummary`.
722 Parameters
723 ----------
724 info : `DatasetInfo`
725 All relevant information on the dataset.
726 producer_info : `QuantumInfo`
727 All relevant information on the producer task. This is used to
728 report the data_id of the producer task.
730 Returns
731 -------
732 summary : `CursedDatasetSummary`
733 A Pydantic model containing the dataID of the task which produced
734 this cursed dataset, the dataID associated with the cursed dataset,
735 run collection names (and their `DatasetRun` information) as well
736 as any messages which may point to any clues about the nature of
737 the problem. These are be messages generated during the assembly of
738 the `QuantumProvenanceGraph` that describe why it was marked as
739 cursed.
740 """
741 runs_visible = {k for k, v in info["runs"].items() if v.visible}
742 return cls(
743 producer_data_id=dict(producer_info["data_id"].required),
744 data_id=dict(info["data_id"].required),
745 runs_produced={k: v.produced for k, v in info["runs"].items()},
746 # this has at most one element
747 run_visible=runs_visible.pop() if runs_visible else None,
748 messages=info["messages"],
749 )
752class DatasetTypeSummary(pydantic.BaseModel):
753 """A summary of the status of all datasets of a particular type across all
754 runs.
755 """
757 producer: str = ""
758 """The name of the task which produced this dataset.
759 """
761 n_visible: int = 0
762 """A count of the datasets of this type which were visible in the
763 finalized collection(s).
764 """
765 n_shadowed: int = 0
766 """A count of the datasets of this type which were produced but not
767 visible. This includes any datasets which do not come up in a butler
768 query over their associated collection.
769 """
770 n_predicted_only: int = 0
771 """A count of the datasets of this type which were predicted but
772 ultimately not produced. Note that this does not indicate a failure,
773 which are accounted for differently. This is commonly referred to as
774 a `NoWorkFound` case.
775 """
776 n_expected: int = 0
777 """The number of datasets of this type expected by the graph.
778 """
780 @pydantic.computed_field # type: ignore[prop-decorator]
781 @property
782 def n_cursed(self) -> int:
783 """Return a count of cursed datasets."""
784 return len(self.cursed_datasets)
786 @pydantic.computed_field # type: ignore[prop-decorator]
787 @property
788 def n_unsuccessful(self) -> int:
789 """Return a count of unsuccessful datasets."""
790 return len(self.unsuccessful_datasets)
792 cursed_datasets: list[CursedDatasetSummary] = pydantic.Field(default_factory=list)
793 """A list of all `CursedDatasetSummary` objects associated with the
794 cursed datasets. This is a report containing their data_ids and the
795 data_ids of their producer task, the status of each run associated with
796 each `cursed` dataset, and messages (dictated in this module) associated
797 with the particular issue identified.
798 """
799 unsuccessful_datasets: list[dict[str, DataIdValue]] = pydantic.Field(default_factory=list)
800 """A list of all unsuccessful datasets by their name and data_id.
801 """
803 def _add_dataset_info(self, info: DatasetInfo, producer_info: QuantumInfo) -> None:
804 """Add a `DatasetInfo` to a `DatasetTypeSummary`.
806 Unpack the `DatasetInfo` object, sorting datasets of each status into
807 the correct place in the `DatasetTypeSummary`. If the status of a
808 dataset is not valid, raise an `AssertionError`.
810 Parameters
811 ----------
812 info : `DatasetInfo`
813 The `DatasetInfo` object to add to the `DatasetTypeSummary`.
814 producer_info : `QuantumInfo`
815 The `QuantumInfo` object associated with the producer of the
816 dataset. This is used to report the producer task in the
817 summaries for cursed datasets, which may help identify
818 specific issues.
819 """
820 match info["status"]:
821 case DatasetInfoStatus.VISIBLE:
822 self.n_visible += 1
823 case DatasetInfoStatus.SHADOWED:
824 self.n_shadowed += 1
825 case DatasetInfoStatus.UNSUCCESSFUL:
826 self.unsuccessful_datasets.append(dict(info["data_id"].mapping))
827 case DatasetInfoStatus.CURSED:
828 self.cursed_datasets.append(CursedDatasetSummary._from_info(info, producer_info))
829 case DatasetInfoStatus.PREDICTED_ONLY:
830 self.n_predicted_only += 1
831 case unrecognized_state:
832 raise AssertionError(f"Unrecognized dataset status {unrecognized_state!r}")
834 def _add_data_id_group(self, other_summary: DatasetTypeSummary) -> None:
835 """Add information from a `DatasetTypeSummary` over one
836 dataquery-identified group to another, as part of aggregating `Summary`
837 reports.
839 Parameters
840 ----------
841 other_summary : `DatasetTypeSummary`
842 `DatasetTypeSummary` to aggregate.
843 """
844 if self.producer and other_summary.producer:
845 # Guard against empty string
846 if self.producer != other_summary.producer:
847 _LOG.warning(
848 "Producer for dataset type is not consistent: %r != %r.",
849 self.producer,
850 other_summary.producer,
851 )
852 _LOG.warning("Ignoring %r.", other_summary.producer)
853 else:
854 if other_summary.producer and not self.producer:
855 self.producer = other_summary.producer
857 self.n_visible += other_summary.n_visible
858 self.n_shadowed += other_summary.n_shadowed
859 self.n_predicted_only += other_summary.n_predicted_only
860 self.n_expected += other_summary.n_expected
862 self.cursed_datasets.extend(other_summary.cursed_datasets)
863 self.unsuccessful_datasets.extend(other_summary.unsuccessful_datasets)
866class Summary(pydantic.BaseModel):
867 """A summary of the contents of the QuantumProvenanceGraph, including
868 all information on the quanta for each task and the datasets of each
869 `DatasetType`.
870 """
872 tasks: dict[str, TaskSummary] = pydantic.Field(default_factory=dict)
873 """Summaries for the tasks and their quanta.
874 """
876 datasets: dict[str, DatasetTypeSummary] = pydantic.Field(default_factory=dict)
877 """Summaries for the datasets.
878 """
880 @classmethod
881 def aggregate(cls, summaries: Sequence[Summary]) -> Summary:
882 """Combine summaries from disjoint data id groups into an overall
883 summary of common tasks and datasets. Intended for use when the same
884 pipeline has been run over all groups.
886 Parameters
887 ----------
888 summaries : `Sequence[Summary]`
889 Sequence of all `Summary` objects to aggregate.
890 """
891 result = cls()
892 for summary in summaries:
893 for label, task_summary in summary.tasks.items():
894 result_task_summary = result.tasks.setdefault(label, TaskSummary())
895 result_task_summary._add_data_id_group(task_summary)
896 for dataset_type, dataset_type_summary in summary.datasets.items():
897 result_dataset_summary = result.datasets.setdefault(dataset_type, DatasetTypeSummary())
898 result_dataset_summary._add_data_id_group(dataset_type_summary)
899 return result
901 def pprint(self, brief: bool = False, datasets: bool = True) -> None:
902 """Print this summary to stdout, as a series of tables.
904 Parameters
905 ----------
906 brief : `bool`, optional
907 If `True`, only display short (counts-only) tables. By default,
908 per-data ID information for exceptions and failures are printed as
909 well.
910 datasets : `bool`, optional
911 Whether to include tables of datasets as well as quanta. This
912 includes a summary table of dataset counts for various status and
913 (if ``brief`` is `True`) a table with per-data ID information for
914 each unsuccessful or cursed dataset.
915 """
916 self.make_quantum_table().pprint_all()
917 print("")
918 print("Caveat codes:")
919 for k, v in QuantumSuccessCaveats.legend().items():
920 print(f"{k}: {v}")
921 print("")
922 if exception_table := self.make_exception_table():
923 exception_table.pprint_all()
924 print("")
925 if datasets:
926 self.make_dataset_table().pprint_all()
927 print("")
928 if not brief:
929 for task_label, bad_quantum_table in self.make_bad_quantum_tables().items():
930 print(f"{task_label} errors:")
931 bad_quantum_table.pprint_all()
932 print("")
933 if datasets:
934 for dataset_type_name, bad_dataset_table in self.make_bad_dataset_tables().items():
935 print(f"{dataset_type_name} errors:")
936 bad_dataset_table.pprint_all()
937 print("")
939 def make_quantum_table(self) -> astropy.table.Table:
940 """Construct an `astropy.table.Table` with a tabular summary of the
941 quanta.
943 Returns
944 -------
945 table : `astropy.table.Table`
946 A table view of the quantum information. This only includes
947 counts of status categories and caveats, not any per-data-ID
948 detail.
950 Notes
951 -----
952 Success caveats in the table are represented by their
953 `~QuantumSuccessCaveats.concise` form, so when pretty-printing this
954 table for users, the `~QuantumSuccessCaveats.legend` should generally
955 be printed as well.
956 """
957 rows = []
958 for label, task_summary in self.tasks.items():
959 if len(task_summary.caveats) > 1:
960 caveats = "(multiple)"
961 elif len(task_summary.caveats) == 1:
962 ((code, data_ids),) = task_summary.caveats.items()
963 caveats = f"{code}({len(data_ids)})"
964 else:
965 caveats = ""
966 rows.append(
967 {
968 "Task": label,
969 "Unknown": task_summary.n_unknown,
970 "Successful": task_summary.n_successful,
971 "Caveats": caveats,
972 "Blocked": task_summary.n_blocked,
973 "Failed": task_summary.n_failed,
974 "Wonky": task_summary.n_wonky,
975 "TOTAL": sum(
976 [
977 task_summary.n_successful,
978 task_summary.n_unknown,
979 task_summary.n_blocked,
980 task_summary.n_failed,
981 task_summary.n_wonky,
982 ]
983 ),
984 "EXPECTED": task_summary.n_expected,
985 }
986 )
987 return astropy.table.Table(rows)
989 def make_dataset_table(self) -> astropy.table.Table:
990 """Construct an `astropy.table.Table` with a tabular summary of the
991 datasets.
993 Returns
994 -------
995 table : `astropy.table.Table`
996 A table view of the dataset information. This only includes
997 counts of status categories, not any per-data-ID detail.
998 """
999 rows = []
1000 for dataset_type_name, dataset_type_summary in self.datasets.items():
1001 rows.append(
1002 {
1003 "Dataset": dataset_type_name,
1004 "Visible": dataset_type_summary.n_visible,
1005 "Shadowed": dataset_type_summary.n_shadowed,
1006 "Predicted Only": dataset_type_summary.n_predicted_only,
1007 "Unsuccessful": dataset_type_summary.n_unsuccessful,
1008 "Cursed": dataset_type_summary.n_cursed,
1009 "TOTAL": sum(
1010 [
1011 dataset_type_summary.n_visible,
1012 dataset_type_summary.n_shadowed,
1013 dataset_type_summary.n_predicted_only,
1014 dataset_type_summary.n_unsuccessful,
1015 dataset_type_summary.n_cursed,
1016 ]
1017 ),
1018 "EXPECTED": dataset_type_summary.n_expected,
1019 }
1020 )
1021 return astropy.table.Table(rows)
1023 def make_exception_table(self) -> astropy.table.Table:
1024 """Construct an `astropy.table.Table` with counts for each exception
1025 type raised by each task.
1027 At present this only includes information from partial-outputs-error
1028 successes, since exception information for failures is not tracked.
1029 This may change in the future.
1031 Returns
1032 -------
1033 table : `astropy.table.Table`
1034 A table with columns for task label, exception type, and counts.
1035 """
1036 rows = []
1037 for task_label, task_summary in self.tasks.items():
1038 for type_name, exception_summaries in task_summary.exceptions.items():
1039 rows.append({"Task": task_label, "Exception": type_name, "Count": len(exception_summaries)})
1040 return astropy.table.Table(rows)
1042 def make_bad_quantum_tables(self, max_message_width: int = 80) -> dict[str, astropy.table.Table]:
1043 """Construct an `astropy.table.Table` with per-data-ID information
1044 about failed, wonky, and partial-outputs-error quanta.
1046 Parameters
1047 ----------
1048 max_message_width : `int`, optional
1049 Maximum width for the Message column. Longer messages are
1050 truncated.
1052 Returns
1053 -------
1054 tables : `dict` [ `str`, `astropy.table.Table` ]
1055 A table for each task with status, data IDs, and log messages for
1056 each unsuccessful quantum. Keys are task labels. Only task with
1057 unsuccessful quanta or partial outputs errors are included.
1058 """
1059 result = {}
1060 for task_label, task_summary in self.tasks.items():
1061 rows = []
1062 for status, unsuccessful_quantum_summary in itertools.chain(
1063 zip(itertools.repeat("FAILED"), task_summary.failed_quanta),
1064 zip(itertools.repeat("WONKY"), task_summary.wonky_quanta),
1065 ):
1066 row = {"Status(Caveats)": status, "Exception": "", **unsuccessful_quantum_summary.data_id}
1067 row["Message"] = (
1068 textwrap.shorten(unsuccessful_quantum_summary.messages[-1], max_message_width)
1069 if unsuccessful_quantum_summary.messages
1070 else ""
1071 )
1072 rows.append(row)
1073 for exception_summary in itertools.chain.from_iterable(task_summary.exceptions.values()):
1074 # Trim off the package name from the exception type for
1075 # brevity.
1076 short_name: str = exception_summary.exception.type_name.rsplit(".", maxsplit=1)[-1]
1077 row = {
1078 "Status(Caveats)": "SUCCESSFUL(P)", # we only get exception info for partial outputs
1079 "Exception": short_name,
1080 **exception_summary.data_id,
1081 "Message": textwrap.shorten(exception_summary.exception.message, max_message_width),
1082 }
1083 rows.append(row)
1084 if rows:
1085 table = astropy.table.Table(rows)
1086 table.columns["Exception"].format = "<"
1087 table.columns["Message"].format = "<"
1088 result[task_label] = table
1089 return result
1091 def make_bad_dataset_tables(self, max_message_width: int = 80) -> dict[str, astropy.table.Table]:
1092 """Construct an `astropy.table.Table` with per-data-ID information
1093 about unsuccessful and cursed datasets.
1095 Parameters
1096 ----------
1097 max_message_width : `int`, optional
1098 Maximum width for the Message column. Longer messages are
1099 truncated.
1101 Returns
1102 -------
1103 tables : `dict` [ `str`, `astropy.table.Table` ]
1104 A table for each task with status, data IDs, and log messages for
1105 each unsuccessful quantum. Keys are task labels. Only task with
1106 unsuccessful quanta are included.
1107 """
1108 result = {}
1109 for dataset_type_name, dataset_type_summary in self.datasets.items():
1110 rows = []
1111 for data_id in dataset_type_summary.unsuccessful_datasets:
1112 row = {"Status": "UNSUCCESSFUL", **data_id, "Message": ""}
1113 for cursed_dataset_summary in dataset_type_summary.cursed_datasets:
1114 row = {"Status": "CURSED", **cursed_dataset_summary.data_id}
1115 row["Message"] = (
1116 textwrap.shorten(cursed_dataset_summary.messages[-1], max_message_width)
1117 if cursed_dataset_summary.messages
1118 else ""
1119 )
1120 rows.append(row)
1121 if rows:
1122 table = astropy.table.Table(rows)
1123 table.columns["Message"].format = "<"
1124 result[dataset_type_name] = table
1125 return result
1128class QuantumProvenanceGraph:
1129 """A set of already-run, merged quantum graphs with provenance
1130 information.
1132 Parameters
1133 ----------
1134 butler : `lsst.daf.butler.Butler`
1135 The Butler used for this report. This should match the Butler used
1136 for the run associated with the executed quantum graph.
1137 qgraphs : `~collections.abc.Sequence` [`QuantumGraph` |\
1138 `~lsst.utils.resources.ResourcePathExpression`]
1139 A list of either quantum graph objects or their uri's, to be used
1140 to assemble the `QuantumProvenanceGraph`.
1141 collections : `~collections.abc.Sequence` [`str`] | `None`
1142 Collections to use in `lsst.daf.butler.query_datasets` when testing
1143 which datasets are available at a high level.
1144 where : `str`
1145 A "where" string to use to constrain the datasets; should be provided
1146 if ``collections`` includes many datasets that are not in any graphs,
1147 to select just those that might be (e.g. when sharding over dimensions
1148 and using a final collection that spans multiple shards).
1149 curse_failed_logs : `bool`
1150 Mark log datasets as CURSED if they are visible in the final output
1151 collection. Note that a campaign-level collection must be used here for
1152 `collections` if `curse_failed_logs` is `True`.
1153 read_caveats : `str` or `None`, optional
1154 Whether to read metadata files to get flags that describe qualified
1155 successes. If `None`, no metadata files will be read and all
1156 ``caveats`` fields will be `None`. If "exhaustive", all metadata files
1157 will be read. If "lazy", only metadata files where at least one
1158 predicted output is missing will be read.
1159 use_qbb : `bool`, optional
1160 If `True`, use a quantum-backed butler when reading metadata files.
1161 Note that some butler database queries are still run even if this is
1162 `True`; this does not avoid database access entirely.
1163 n_cores : `int`, optional
1164 Number of threads to use for parallelization.
1165 """
1167 def __init__(
1168 self,
1169 butler: Butler | None = None,
1170 qgraphs: Sequence[QuantumGraph | ResourcePathExpression] = (),
1171 *,
1172 collections: Sequence[str] | None = None,
1173 where: str = "",
1174 curse_failed_logs: bool = False,
1175 read_caveats: Literal["lazy", "exhaustive"] | None = "lazy",
1176 use_qbb: bool = True,
1177 n_cores: int = 1,
1178 ) -> None:
1179 # The graph we annotate as we step through all the graphs associated
1180 # with the processing to create the `QuantumProvenanceGraph`.
1181 self._xgraph = networkx.DiGraph()
1182 # The nodes representing quanta in `_xgraph` grouped by task label.
1183 self._quanta: dict[str, set[QuantumKey]] = {}
1184 # The nodes representing datasets in `_xgraph` grouped by dataset type
1185 # name.
1186 self._datasets: dict[str, set[DatasetKey]] = {}
1187 # Bool representing whether the graph has been finalized. This is set
1188 # to True when resolve_duplicates completes.
1189 self._finalized: bool = False
1190 # In order to both parallelize metadata/log reads and potentially use
1191 # QBB to do it, we in general need one butler for each output_run and
1192 # thread combination. This dict is keyed by the former, and the
1193 # wrapper type used for the value handles the latter.
1194 self._butler_wrappers: dict[str, _ThreadLocalButlerWrapper] = {}
1195 if butler is not None:
1196 self.assemble_quantum_provenance_graph(
1197 butler,
1198 qgraphs,
1199 collections=collections,
1200 where=where,
1201 curse_failed_logs=curse_failed_logs,
1202 read_caveats=read_caveats,
1203 use_qbb=use_qbb,
1204 n_cores=n_cores,
1205 )
1206 elif qgraphs:
1207 raise TypeError("'butler' must be provided if `qgraphs` is.")
1209 @property
1210 def quanta(self) -> Mapping[str, Set[QuantumKey]]:
1211 """A mapping from task label to a set of keys for its quanta."""
1212 return self._quanta
1214 @property
1215 def datasets(self) -> Mapping[str, Set[DatasetKey]]:
1216 """A mapping from dataset type name to a set of keys for datasets."""
1217 return self._datasets
1219 def get_quantum_info(self, key: QuantumKey) -> QuantumInfo:
1220 """Get a `QuantumInfo` object from the `QuantumProvenanceGraph` using
1221 a `QuantumKey`.
1223 Parameters
1224 ----------
1225 key : `QuantumKey`
1226 The key used to refer to the node on the graph.
1228 Returns
1229 -------
1230 quantum_info : `QuantumInfo`
1231 The `TypedDict` with information on the task label-dataID pair
1232 across all runs.
1233 """
1234 return self._xgraph.nodes[key]
1236 def get_dataset_info(self, key: DatasetKey) -> DatasetInfo:
1237 """Get a `DatasetInfo` object from the `QuantumProvenanceGraph` using
1238 a `DatasetKey`.
1240 Parameters
1241 ----------
1242 key : `DatasetKey`
1243 The key used to refer to the node on the graph.
1245 Returns
1246 -------
1247 dataset_info : `DatasetInfo`
1248 The `TypedDict` with information about the `DatasetType`-dataID
1249 pair across all runs.
1250 """
1251 return self._xgraph.nodes[key]
1253 def to_summary(
1254 self, butler: Butler | None = None, do_store_logs: bool = True, n_cores: int = 1
1255 ) -> Summary:
1256 """Summarize the `QuantumProvenanceGraph`.
1258 Parameters
1259 ----------
1260 butler : `lsst.daf.butler.Butler`, optional
1261 Ignored; accepted for backwards compatibility.
1262 do_store_logs : `bool`
1263 Store the logs in the summary dictionary.
1264 n_cores : `int`, optional
1266 Returns
1267 -------
1268 result : `Summary`
1269 A struct containing counts of quanta and datasets in each of
1270 the overall states defined in `QuantumInfo` and `DatasetInfo`,
1271 as well as diagnostic information and error messages for failed
1272 quanta and strange edge cases, and a list of recovered quanta.
1273 """
1274 status_log = PeriodicLogger(_LOG)
1275 if not self._finalized:
1276 raise RuntimeError(
1277 """resolve_duplicates must be called to finalize the
1278 QuantumProvenanceGraph before making a summary."""
1279 )
1280 result = Summary()
1281 futures: list[concurrent.futures.Future[None]] = []
1282 _LOG.verbose("Summarizing %s tasks.", len(self._quanta.keys()))
1283 with concurrent.futures.ThreadPoolExecutor(n_cores) as executor:
1284 for m, (task_label, quanta) in enumerate(self._quanta.items()):
1285 task_summary = TaskSummary()
1286 task_summary.n_expected = len(quanta)
1287 for n, quantum_key in enumerate(quanta):
1288 quantum_info = self.get_quantum_info(quantum_key)
1289 future = task_summary._add_quantum_info(
1290 quantum_info,
1291 log_getter=self._butler_get if do_store_logs else None,
1292 executor=executor,
1293 )
1294 if future is not None:
1295 futures.append(future)
1296 status_log.log(
1297 "Summarized %s of %s quanta of task %s of %s.",
1298 n + 1,
1299 len(quanta),
1300 m + 1,
1301 len(self._quanta.keys()),
1302 )
1303 result.tasks[task_label] = task_summary
1304 for n, future in enumerate(concurrent.futures.as_completed(futures)):
1305 if (err := future.exception()) is not None:
1306 raise err
1307 status_log.log("Loaded messages from %s of %s log datasets.", n + 1, len(futures))
1308 _LOG.verbose("Summarizing %s dataset types.", len(self._datasets.keys()))
1309 for m, (dataset_type_name, datasets) in enumerate(self._datasets.items()):
1310 dataset_type_summary = DatasetTypeSummary(producer="")
1311 dataset_type_summary.n_expected = len(datasets)
1312 for n, dataset_key in enumerate(datasets):
1313 dataset_info = self.get_dataset_info(dataset_key)
1314 producer_key = self.get_producer_of(dataset_key)
1315 producer_info = self.get_quantum_info(producer_key)
1316 # Not ideal, but hard to get out of the graph at the moment.
1317 # Change after DM-40441
1318 dataset_type_summary.producer = producer_key.task_label
1319 dataset_type_summary._add_dataset_info(dataset_info, producer_info)
1320 status_log.log(
1321 "Summarized %s of %s datasets of type %s of %s.",
1322 n + 1,
1323 len(datasets),
1324 m + 1,
1325 len(self._datasets.keys()),
1326 )
1327 result.datasets[dataset_type_name] = dataset_type_summary
1328 return result
1330 def iter_outputs_of(self, quantum_key: QuantumKey) -> Iterator[DatasetKey]:
1331 """Iterate through the outputs of a quantum, yielding the keys of
1332 all of the datasets produced by the quantum.
1334 Parameters
1335 ----------
1336 quantum_key : `QuantumKey`
1337 The key for the quantum whose outputs are needed.
1338 """
1339 yield from self._xgraph.successors(quantum_key)
1341 def get_producer_of(self, dataset_key: DatasetKey) -> QuantumKey:
1342 """Unpack the predecessor (producer quantum) of a given dataset key
1343 from a graph.
1345 Parameters
1346 ----------
1347 dataset_key : `DatasetKey`
1348 The key for the dataset whose producer quantum is needed.
1350 Returns
1351 -------
1352 result : `QuantumKey`
1353 The key for the quantum which produced the dataset.
1354 """
1355 (result,) = self._xgraph.predecessors(dataset_key)
1356 return result
1358 def iter_downstream(
1359 self, key: QuantumKey | DatasetKey
1360 ) -> Iterator[tuple[QuantumKey, QuantumInfo] | tuple[DatasetKey, DatasetInfo]]:
1361 """Iterate over the quanta and datasets that are downstream of a
1362 quantum or dataset.
1364 Parameters
1365 ----------
1366 key : `QuantumKey` or `DatasetKey`
1367 Starting node.
1369 Returns
1370 -------
1371 iter : `~collections.abc.Iterator` [ `tuple` ]
1372 An iterator over pairs of (`QuantumKey`, `QuantumInfo`) or
1373 (`DatasetKey`, `DatasetInfo`).
1374 """
1375 for key in networkx.dag.descendants(self._xgraph, key):
1376 yield (key, self._xgraph.nodes[key]) # type: ignore
1378 def assemble_quantum_provenance_graph(
1379 self,
1380 butler: Butler,
1381 qgraphs: Sequence[QuantumGraph | ResourcePathExpression],
1382 collections: Sequence[str] | None = None,
1383 where: str = "",
1384 curse_failed_logs: bool = False,
1385 read_caveats: Literal["lazy", "exhaustive"] | None = "lazy",
1386 use_qbb: bool = True,
1387 n_cores: int = 1,
1388 ) -> None:
1389 """Assemble the quantum provenance graph from a list of all graphs
1390 corresponding to processing attempts.
1392 Parameters
1393 ----------
1394 butler : `lsst.daf.butler.Butler`
1395 The Butler used for this report. This should match the Butler used
1396 for the run associated with the executed quantum graph.
1397 qgraphs : `~collections.abc.Sequence` [`QuantumGraph` |\
1398 `~lsst.utils.resources.ResourcePathExpression`]
1399 A list of either quantum graph objects or their uri's, to be used
1400 to assemble the `QuantumProvenanceGraph`.
1401 collections : `~collections.abc.Sequence` [`str`] | `None`
1402 Collections to use in `lsst.daf.butler.query_datasets` when testing
1403 which datasets are available at a high level.
1404 where : `str`
1405 A "where" string to use to constrain the datasets; should be
1406 provided if ``collections`` includes many datasets that are not in
1407 any graphs, to select just those that might be (e.g. when sharding
1408 over dimensions and using a final collection that spans multiple
1409 shards).
1410 curse_failed_logs : `bool`
1411 Mark log datasets as CURSED if they are visible in the final
1412 output collection. Note that a campaign-level collection must be
1413 used here for `collections` if `curse_failed_logs` is `True`.
1414 read_caveats : `str` or `None`, optional
1415 Whether to read metadata files to get flags that describe qualified
1416 successes. If `None`, no metadata files will be read and all
1417 ``caveats`` fields will be `None`. If "exhaustive", all
1418 metadata files will be read. If "lazy", only metadata files where
1419 at least one predicted output is missing will be read.
1420 use_qbb : `bool`, optional
1421 If `True`, use a quantum-backed butler when reading metadata files.
1422 Note that some butler database queries are still run even if this
1423 is `True`; this does not avoid database access entirely.
1424 n_cores : `int`, optional
1425 Number of threads to use for parallelization.
1426 """
1427 if read_caveats not in ("lazy", "exhaustive", None):
1428 raise TypeError(
1429 f"Invalid option {read_caveats!r} for read_caveats; should be 'lazy', 'exhaustive', or None."
1430 )
1431 output_runs = []
1432 last_time: datetime.datetime | None = None
1433 for graph in qgraphs:
1434 if not isinstance(graph, QuantumGraph):
1435 _LOG.verbose("Loading quantum graph %r.", graph)
1436 qgraph = QuantumGraph.loadUri(graph)
1437 else:
1438 qgraph = graph
1439 assert qgraph.metadata is not None, "Saved QGs always have metadata."
1440 self._add_new_graph(butler, qgraph, read_caveats=read_caveats, use_qbb=use_qbb, n_cores=n_cores)
1441 output_runs.append(qgraph.metadata["output_run"])
1442 if last_time is not None and last_time > qgraph.metadata["time"]:
1443 raise RuntimeError("Quantum graphs must be passed in chronological order.")
1444 last_time = qgraph.metadata["time"]
1445 if not collections:
1446 # We reverse the order of the associated output runs because the
1447 # query in _resolve_duplicates must be done most-recent first.
1448 collections = list(reversed(output_runs))
1449 assert not curse_failed_logs, (
1450 "curse_failed_logs option must be used with one campaign-level collection."
1451 )
1452 self._resolve_duplicates(butler, collections, where, curse_failed_logs)
1454 def _add_new_graph(
1455 self,
1456 butler: Butler,
1457 qgraph: QuantumGraph,
1458 read_caveats: Literal["lazy", "exhaustive"] | None,
1459 use_qbb: bool = True,
1460 n_cores: int = 1,
1461 ) -> None:
1462 """Add a new quantum graph to the `QuantumProvenanceGraph`.
1464 Parameters
1465 ----------
1466 butler : `lsst.daf.butler.Butler`
1467 The Butler used for this report. This should match the Butler
1468 used for the run associated with the executed quantum graph.
1469 qgraph : `QuantumGraph`
1470 The quantum graph object to add.
1471 read_caveats : `str` or `None`
1472 Whether to read metadata files to get flags that describe qualified
1473 successes. If `None`, no metadata files will be read and all
1474 ``caveats`` fields will be `None`. If "exhaustive", all
1475 metadata files will be read. If "lazy", only metadata files where
1476 at least one predicted output is missing will be read.
1477 use_qbb : `bool`, optional
1478 If `True`, use a quantum-backed butler when reading metadata files.
1479 Note that some butler database queries are still run even if this
1480 is `True`; this does not avoid database access entirely.
1481 n_cores : `int`, optional
1482 Number of threads to use for parallelization.
1483 """
1484 status_log = PeriodicLogger(_LOG)
1485 output_run = qgraph.metadata["output_run"]
1486 # Add QuantumRun and DatasetRun (and nodes/edges, as needed) to the
1487 # QPG for all quanta in the QG.
1488 _LOG.verbose("Adding output run to provenance graph.")
1489 new_quanta: list[QuantumKey] = []
1490 for n, node in enumerate(qgraph):
1491 new_quanta.append(self._add_new_quantum(node, output_run))
1492 status_log.log("Added nodes for %s of %s quanta.", n + 1, len(qgraph))
1493 # Query for datasets in the output run to see which ones were actually
1494 # produced.
1495 _LOG.verbose("Querying for existence for %s dataset types.", len(self._datasets.keys()))
1496 for m, dataset_type_name in enumerate(self._datasets):
1497 try:
1498 refs = butler.query_datasets(
1499 dataset_type_name, collections=output_run, explain=False, limit=None
1500 )
1501 except MissingDatasetTypeError:
1502 continue
1503 for n, ref in enumerate(refs):
1504 dataset_key = DatasetKey(ref.datasetType.name, ref.dataId.required_values)
1505 dataset_info = self.get_dataset_info(dataset_key)
1506 dataset_run = dataset_info["runs"][output_run] # dataset run (singular)
1507 dataset_run.produced = True
1508 status_log.log(
1509 "Updated status for %s of %s datasets of %s of %s types.",
1510 n + 1,
1511 len(refs),
1512 m + 1,
1513 len(self._datasets.keys()),
1514 )
1515 if use_qbb:
1516 _LOG.verbose("Using quantum-backed butler for metadata loads.")
1517 self._butler_wrappers[output_run] = _ThreadLocalButlerWrapper.wrap_qbb(butler, qgraph)
1518 else:
1519 _LOG.verbose("Using full butler for metadata loads.")
1520 self._butler_wrappers[output_run] = _ThreadLocalButlerWrapper.wrap_full(butler)
1522 _LOG.verbose("Setting quantum status from dataset existence.")
1523 # Update quantum status information based on which datasets were
1524 # produced.
1525 blocked: set[DatasetKey] = set() # the outputs of failed or blocked quanta in this run.
1526 with concurrent.futures.ThreadPoolExecutor(n_cores) as executor:
1527 futures: list[concurrent.futures.Future[None]] = []
1528 for n, quantum_key in enumerate(new_quanta):
1529 if (
1530 self._update_run_status(quantum_key, output_run, blocked) == QuantumRunStatus.SUCCESSFUL
1531 and read_caveats is not None
1532 ):
1533 self._update_caveats(quantum_key, output_run, read_caveats, executor, futures)
1534 self._update_info_status(quantum_key, output_run)
1535 status_log.log("Updated status for %s of %s quanta.", n + 1, len(new_quanta))
1536 for n, future in enumerate(concurrent.futures.as_completed(futures)):
1537 if (err := future.exception()) is not None:
1538 raise err
1539 status_log.log("Added exception/caveat information for %s of %s quanta.", n + 1, len(futures))
1541 def _add_new_quantum(self, node: QuantumNode, output_run: str) -> QuantumKey:
1542 """Add a quantum from a new quantum graph to the provenance graph.
1544 Parameters
1545 ----------
1546 node : `QuantumNode`
1547 Node in the quantum graph.
1548 output_run : `str`
1549 Output run collection.
1551 Returns
1552 -------
1553 quantum_key : `QuantumKey`
1554 Key for the new or existing node in the provenance graph.
1556 Notes
1557 -----
1558 This method adds new quantum and dataset nodes to the provenance graph
1559 if they don't already exist, while adding new `QuantumRun` and
1560 `DatasetRun` objects to both new and existing nodes. All status
1561 information on those nodes is set to initial, default values that
1562 generally reflect quanta that have not been attempted to be run.
1563 """
1564 # make a key to refer to the quantum and add it to the quantum
1565 # provenance graph.
1566 quantum_key = QuantumKey(
1567 node.taskDef.label, cast(DataCoordinate, node.quantum.dataId).required_values
1568 )
1569 self._xgraph.add_node(quantum_key)
1570 # use the key to get a `QuantumInfo` object for the quantum
1571 # and set defaults for its values.
1572 quantum_info = self.get_quantum_info(quantum_key)
1573 quantum_info.setdefault("messages", [])
1574 quantum_info.setdefault("runs", {})
1575 quantum_info.setdefault("data_id", cast(DataCoordinate, node.quantum.dataId))
1576 quantum_info.setdefault("status", QuantumInfoStatus.UNKNOWN)
1577 quantum_info.setdefault("recovered", False)
1578 self._quanta.setdefault(quantum_key.task_label, set()).add(quantum_key)
1579 metadata_ref = node.quantum.outputs[METADATA_OUTPUT_TEMPLATE.format(label=node.taskDef.label)][0]
1580 log_ref = node.quantum.outputs[LOG_OUTPUT_TEMPLATE.format(label=node.taskDef.label)][0]
1581 # associate run collections with specific quanta. this is important
1582 # if the same quanta are processed in multiple runs as in recovery
1583 # workflows.
1584 quantum_runs = quantum_info.setdefault("runs", {})
1585 # the `QuantumRun` here is the specific quantum-run collection
1586 # combination.
1587 quantum_runs[output_run] = QuantumRun(id=node.nodeId, metadata_ref=metadata_ref, log_ref=log_ref)
1588 # For each of the outputs of the quanta (datasets) make a key to
1589 # refer to the dataset.
1590 for ref in itertools.chain.from_iterable(node.quantum.outputs.values()):
1591 dataset_key = DatasetKey(ref.datasetType.name, ref.dataId.required_values)
1592 # add datasets to the nodes of the graph, with edges on the
1593 # quanta.
1594 self._xgraph.add_edge(quantum_key, dataset_key)
1595 # use the dataset key to make a `DatasetInfo` object for
1596 # the dataset and set defaults for its values.
1597 dataset_info = self.get_dataset_info(dataset_key)
1598 dataset_info.setdefault("data_id", ref.dataId)
1599 dataset_info.setdefault("status", DatasetInfoStatus.PREDICTED_ONLY)
1600 dataset_info.setdefault("messages", [])
1601 self._datasets.setdefault(dataset_key.dataset_type_name, set()).add(dataset_key)
1602 dataset_runs = dataset_info.setdefault("runs", {})
1603 # make a `DatasetRun` for the specific dataset-run
1604 # collection combination.
1605 dataset_runs[output_run] = DatasetRun(id=ref.id)
1606 # save metadata and logs for easier status interpretation later
1607 if dataset_key.dataset_type_name.endswith(METADATA_OUTPUT_CONNECTION_NAME):
1608 quantum_info["metadata"] = dataset_key
1609 quantum_runs[output_run].metadata_ref = ref
1610 if dataset_key.dataset_type_name.endswith(LOG_OUTPUT_CONNECTION_NAME):
1611 quantum_info["log"] = dataset_key
1612 quantum_runs[output_run].log_ref = ref
1613 for ref in itertools.chain.from_iterable(node.quantum.inputs.values()):
1614 dataset_key = DatasetKey(ref.datasetType.nameAndComponent()[0], ref.dataId.required_values)
1615 if dataset_key in self._xgraph:
1616 # add another edge if the input datasetType and quantum are
1617 # in the graph
1618 self._xgraph.add_edge(dataset_key, quantum_key)
1619 return quantum_key
1621 def _update_run_status(
1622 self, quantum_key: QuantumKey, output_run: str, blocked: set[DatasetKey]
1623 ) -> QuantumRunStatus:
1624 """Update the status of this quantum in its own output run, using
1625 information in the graph about which of its output datasets exist.
1627 Parameters
1628 ----------
1629 quantum_key : `QuantumKey`
1630 Key for the node in the provenance graph.
1631 output_run : `str`
1632 Output run collection.
1633 blocked : `set` [ `DatasetKey` ]
1634 A set of output datasets (for all quanta, not just this one) that
1635 were blocked by failures. Will be modified in place.
1637 Returns
1638 -------
1639 run_status : `QuantumRunStatus`
1640 Run-specific status for this quantum.
1641 """
1642 quantum_info = self.get_quantum_info(quantum_key)
1643 quantum_run = quantum_info["runs"][output_run]
1644 metadata_key = quantum_info["metadata"]
1645 log_key = quantum_info["log"]
1646 metadata_dataset_run = self.get_dataset_info(metadata_key)["runs"][output_run]
1647 log_dataset_run = self.get_dataset_info(log_key)["runs"][output_run]
1648 # if we do have metadata, we know that the task finished.
1649 if metadata_dataset_run.produced:
1650 # if we also have logs, this is a success.
1651 if log_dataset_run.produced:
1652 quantum_run.status = QuantumRunStatus.SUCCESSFUL
1653 else:
1654 # if we have metadata and no logs, this is a very rare
1655 # case. either the task ran successfully and the datastore
1656 # died immediately afterwards, or some supporting
1657 # infrastructure for transferring the logs to the datastore
1658 # failed.
1659 quantum_run.status = QuantumRunStatus.LOGS_MISSING
1661 # missing metadata means that the task did not finish.
1662 else:
1663 # if we have logs and no metadata, the task not finishing is
1664 # a failure in the task itself. This includes all payload
1665 # errors and some other problems.
1666 if log_dataset_run.produced:
1667 quantum_run.status = QuantumRunStatus.FAILED
1668 # if a quantum fails, all its successor datasets are
1669 # blocked.
1670 blocked.update(self._xgraph.successors(quantum_key))
1671 # if we are missing metadata and logs, either the task was not
1672 # started, or a hard external environmental error prevented
1673 # it from writing logs or metadata.
1674 else:
1675 # if none of this quantum's inputs were blocked, the
1676 # metadata must just be missing.
1677 if blocked.isdisjoint(self._xgraph.predecessors(quantum_key)):
1678 # None of this quantum's inputs were blocked.
1679 quantum_run.status = QuantumRunStatus.METADATA_MISSING
1680 # otherwise we can assume from no metadata and no logs
1681 # that the task was blocked by an upstream failure.
1682 else:
1683 quantum_run.status = QuantumRunStatus.BLOCKED
1684 blocked.update(self._xgraph.successors(quantum_key))
1685 return quantum_run.status
1687 def _update_info_status(self, quantum_key: QuantumKey, output_run: str) -> QuantumInfoStatus:
1688 """Update the status of this quantum across all runs with the status
1689 for its latest run.
1691 Parameters
1692 ----------
1693 quantum_key : `QuantumKey`
1694 Key for the node in the provenance graph.
1695 output_run : `str`
1696 Output run collection.
1698 Returns
1699 -------
1700 info_status : `QuantumRunStatus`
1701 Run-specific status for this quantum.
1702 """
1703 # Now we can start using state transitions to mark overall status.
1704 quantum_info = self.get_quantum_info(quantum_key)
1705 quantum_run = quantum_info["runs"][output_run]
1706 last_status = quantum_info["status"]
1707 new_status: QuantumInfoStatus
1708 match last_status, quantum_run.status:
1709 # A quantum can never escape a WONKY state.
1710 case (QuantumInfoStatus.WONKY, _):
1711 new_status = QuantumInfoStatus.WONKY
1712 # Any transition to a success (excluding from WONKY) is
1713 # a success; any transition from a failed state is also a
1714 # recovery.
1715 case (_, QuantumRunStatus.SUCCESSFUL):
1716 new_status = QuantumInfoStatus.SUCCESSFUL
1717 if last_status != QuantumInfoStatus.SUCCESSFUL and last_status != QuantumInfoStatus.UNKNOWN:
1718 quantum_info["recovered"] = True
1719 # Missing logs are one of the categories of wonky quanta. They
1720 # interfere with our ability to discern quantum status and are
1721 # signs of weird things afoot in processing. Add a message
1722 # noting why this quantum is being marked as wonky to be stored
1723 # in its `UnsuccessfulQuantumInfo`.
1724 case (_, QuantumRunStatus.LOGS_MISSING):
1725 new_status = QuantumInfoStatus.WONKY
1726 quantum_info["messages"].append(f"Logs missing for run {output_run!r}.")
1727 # Leaving a successful state is another category of wonky
1728 # quanta. If a previous success fails on a subsequent run,
1729 # a human should inspect why. Add a message noting why this
1730 # quantum is being marked as wonky to be stored in its
1731 # `UnsuccessfulQuantumInfo`.
1732 case (QuantumInfoStatus.SUCCESSFUL, _):
1733 new_status = QuantumInfoStatus.WONKY
1734 quantum_info["messages"].append(
1735 f"Status went from successful in run {list(quantum_info['runs'].values())[-1]!r} "
1736 f"to {quantum_run.status!r} in {output_run!r}."
1737 )
1738 # If a quantum status is unknown and it moves to blocked, we
1739 # know for sure that it is a blocked quantum.
1740 case (QuantumInfoStatus.UNKNOWN, QuantumRunStatus.BLOCKED):
1741 new_status = QuantumInfoStatus.BLOCKED
1742 # A transition into blocked does not change the overall quantum
1743 # status for a failure.
1744 case (_, QuantumRunStatus.BLOCKED):
1745 new_status = last_status
1746 # If a quantum transitions from any state into missing
1747 # metadata, we don't have enough information to diagnose its
1748 # state.
1749 case (_, QuantumRunStatus.METADATA_MISSING):
1750 new_status = QuantumInfoStatus.UNKNOWN
1751 # Any transition into failure is a failed state.
1752 case (_, QuantumRunStatus.FAILED):
1753 new_status = QuantumInfoStatus.FAILED
1754 # Update `QuantumInfo.status` for this quantum.
1755 quantum_info["status"] = new_status
1756 return new_status
1758 def _update_caveats(
1759 self,
1760 quantum_key: QuantumKey,
1761 output_run: str,
1762 read_caveats: Literal["lazy", "exhaustive"],
1763 executor: concurrent.futures.Executor,
1764 futures: list[concurrent.futures.Future[None]],
1765 ) -> None:
1766 """Read quantum success caveats and exception information from task
1767 metadata.
1769 Parameters
1770 ----------
1771 quantum_key : `QuantumKey`
1772 Key for the node in the provenance graph.
1773 output_run : `str`
1774 Output run collection.
1775 read_caveats : `str`
1776 Whether to read metadata files to get flags that describe qualified
1777 successes. If "exhaustive", all metadata files will be read. If
1778 "lazy", only metadata files where at least one predicted output is
1779 missing will be read.
1780 butler : `lsst.daf.butler.Butler`
1781 The Butler used for this report. This should match the Butler
1782 used for the run associated with the executed quantum graph.
1783 """
1784 if read_caveats == "lazy" and all(
1785 self.get_dataset_info(dataset_key)["runs"][output_run].produced
1786 for dataset_key in self._xgraph.successors(quantum_key)
1787 ):
1788 return
1789 quantum_info = self.get_quantum_info(quantum_key)
1790 quantum_run = quantum_info["runs"][output_run]
1792 def read_metadata() -> None:
1793 md = self._butler_get(quantum_run.metadata_ref, storageClass=METADATA_OUTPUT_STORAGE_CLASS)
1794 try:
1795 # Int conversion guards against spurious conversion to
1796 # float that can apparently sometimes happen in
1797 # TaskMetadata.
1798 quantum_run.caveats = QuantumSuccessCaveats(int(md["quantum"]["caveats"]))
1799 except LookupError:
1800 pass
1801 try:
1802 quantum_run.exception = ExceptionInfo._from_metadata(md[quantum_key.task_label]["failure"])
1803 except LookupError:
1804 pass
1806 futures.append(executor.submit(read_metadata))
1808 def _resolve_duplicates(
1809 self,
1810 butler: Butler,
1811 collections: Sequence[str] | None = None,
1812 where: str = "",
1813 curse_failed_logs: bool = False,
1814 ) -> None:
1815 """After quantum graphs associated with each run have been added
1816 to the `QuantumProvenanceGraph, resolve any discrepancies between
1817 them and use all attempts to finalize overall status.
1819 Particularly, use the state of each `DatasetRun` in combination with
1820 overall quantum status to ascertain the status of each dataset.
1821 Additionally, if there are multiple visible runs associated with a
1822 dataset, mark the producer quantum as WONKY.
1824 This method should be called after
1825 `QuantumProvenanceGraph._add_new_graph` has been called on every graph
1826 associated with the data processing.
1828 Parameters
1829 ----------
1830 butler : `lsst.daf.butler.Butler`
1831 The Butler used for this report. This should match the Butler used
1832 for the run associated with the executed quantum graph.
1833 collections : `~collections.abc.Sequence` [`str`] | `None`
1834 Collections to use in `lsst.daf.butler.query_datasets` when testing
1835 which datasets are available at a high level.
1836 where : `str`
1837 A "where" string to use to constrain the datasets; should be
1838 provided if ``collections`` includes many datasets that are not in
1839 any graphs, to select just those that might be (e.g. when sharding
1840 over dimensions and using a final collection that spans multiple
1841 shards).
1842 curse_failed_logs : `bool`
1843 Mark log datasets as CURSED if they are visible in the final
1844 output collection. Note that a campaign-level collection must be
1845 used here for `collections` if `curse_failed_logs` is `True`; if
1846 `_resolve_duplicates` is run on a list of group-level collections
1847 then each will only show log datasets from their own failures as
1848 visible and datasets from others will be marked as cursed.
1849 """
1850 # First thing: raise an error if resolve_duplicates has been run
1851 # before on this qpg.
1852 if self._finalized:
1853 raise RuntimeError(
1854 """resolve_duplicates may only be called on a
1855 QuantumProvenanceGraph once. Call only after all graphs have
1856 been added, or make a new graph with all constituent
1857 attempts."""
1858 )
1859 status_log = PeriodicLogger(_LOG)
1860 _LOG.verbose("Querying for dataset visibility.")
1861 for m, dataset_type_name in enumerate(self._datasets):
1862 # find datasets in a larger collection.
1863 try:
1864 refs = butler.query_datasets(
1865 dataset_type_name, collections=collections, where=where, limit=None, explain=False
1866 )
1867 except MissingDatasetTypeError:
1868 continue
1869 for n, ref in enumerate(refs):
1870 dataset_key = DatasetKey(ref.datasetType.name, ref.dataId.required_values)
1871 try:
1872 dataset_info = self.get_dataset_info(dataset_key)
1873 # Ignore if we don't actually have the dataset in any of the
1874 # graphs given.
1875 except KeyError:
1876 continue
1877 # queryable datasets are `visible`.
1878 dataset_info["runs"][ref.run].visible = True
1879 status_log.log(
1880 "Updated visibility for %s of %s datasets of type %s of %s.",
1881 n + 1,
1882 len(refs),
1883 m + 1,
1884 len(self._datasets.keys()),
1885 )
1886 _LOG.verbose("Updating task status from dataset visibility.")
1887 for m, task_quanta in enumerate(self._quanta.values()):
1888 for n, quantum_key in enumerate(task_quanta):
1889 # runs associated with visible datasets.
1890 visible_runs: set[str] = set()
1891 quantum_info = self.get_quantum_info(quantum_key)
1892 # Loop over each dataset in the outputs of a single quantum.
1893 for dataset_key in self.iter_outputs_of(quantum_key):
1894 dataset_info = self.get_dataset_info(dataset_key)
1895 dataset_type_name = dataset_key.dataset_type_name
1896 visible_runs.update(
1897 run for run, dataset_run in dataset_info["runs"].items() if dataset_run.visible
1898 )
1899 if any(dataset_run.visible for dataset_run in dataset_info["runs"].values()):
1900 query_state = "visible"
1901 # set the publish state to `shadowed` if the dataset was
1902 # produced but not visible (i.e., not queryable from the
1903 # final collection(s)).
1904 elif any(dataset_run.produced for dataset_run in dataset_info["runs"].values()):
1905 query_state = "shadowed"
1906 # a dataset which was not produced and not visible is
1907 # missing.
1908 else:
1909 query_state = "missing"
1910 # use the quantum status and publish state to ascertain the
1911 # status of the dataset.
1912 match (quantum_info["status"], query_state):
1913 # visible datasets from successful quanta are as
1914 # intended.
1915 case (QuantumInfoStatus.SUCCESSFUL, "visible"):
1916 dataset_info["status"] = DatasetInfoStatus.VISIBLE
1917 # missing datasets from successful quanta indicate a
1918 # `NoWorkFound` case.
1919 case (QuantumInfoStatus.SUCCESSFUL, "missing"):
1920 dataset_info["status"] = DatasetInfoStatus.PREDICTED_ONLY
1921 case (QuantumInfoStatus.SUCCESSFUL, "shadowed"):
1922 dataset_info["status"] = DatasetInfoStatus.SHADOWED
1923 # If anything other than a successful quantum produces
1924 # a visible dataset, that dataset is cursed. Set the
1925 # status for the dataset to cursed and note the reason
1926 # for labeling the dataset as cursed.
1927 case (_, "visible"):
1928 # Avoiding publishing failed logs is difficult
1929 # without using tagged collections, so flag them as
1930 # merely unsuccessful unless the user requests it.
1931 if (
1932 dataset_type_name.endswith(LOG_OUTPUT_CONNECTION_NAME)
1933 and not curse_failed_logs
1934 ):
1935 dataset_info["status"] = DatasetInfoStatus.UNSUCCESSFUL
1936 else:
1937 dataset_info["status"] = DatasetInfoStatus.CURSED
1938 dataset_info["messages"].append(
1939 f"Unsuccessful dataset {dataset_type_name} visible in "
1940 "final output collection."
1941 )
1942 # any other produced dataset (produced but not
1943 # visible and not successful) is a regular
1944 # failure.
1945 case _:
1946 dataset_info["status"] = DatasetInfoStatus.UNSUCCESSFUL
1947 if len(visible_runs) > 1:
1948 quantum_info["status"] = QuantumInfoStatus.WONKY
1949 quantum_info["messages"].append(
1950 f"Outputs from different runs of the same quanta were visible: {visible_runs}."
1951 )
1952 for dataset_key in self.iter_outputs_of(quantum_key):
1953 dataset_info = self.get_dataset_info(dataset_key)
1954 quantum_info["messages"].append(
1955 f"{dataset_key.dataset_type_name}"
1956 + f"from {str(dataset_info['runs'])};"
1957 + f"{str(dataset_info['status'])}"
1958 )
1959 status_log.log(
1960 "Updated task status from visibility for %s of %s quanta of task %s of %s.",
1961 n + 1,
1962 len(task_quanta),
1963 m + 1,
1964 len(self._quanta.keys()),
1965 )
1966 # If we make it all the way through resolve_duplicates, set
1967 # self._finalized = True so that it cannot be run again.
1968 self._finalized = True
1970 def _butler_get(self, ref: DatasetRef, **kwargs: Any) -> Any:
1971 return self._butler_wrappers[ref.run].butler.get(ref, **kwargs)
1974class _ThreadLocalButlerWrapper:
1975 """A wrapper for a thread-local limited butler.
1977 Parameter
1978 ---------
1979 factory : `~collections.abc.Callable`
1980 A callable that takes no arguments and returns a limited butler.
1981 """
1983 def __init__(self, factory: Callable[[], LimitedButler]):
1984 self._factory = factory
1985 self._thread_local = threading.local()
1987 @classmethod
1988 def wrap_qbb(cls, full_butler: Butler, qg: QuantumGraph) -> _ThreadLocalButlerWrapper:
1989 """Wrap a `~lsst.daf.butler.QuantumBackedButler` suitable for reading
1990 log and metadata files.
1992 Parameters
1993 ----------
1994 full_butler : `~lsst.daf.butler.Butler`
1995 Full butler to draw datastore and dimension configuration from.
1996 qg : `QuantumGraph`
1997 Quantum graph,
1999 Returns
2000 -------
2001 wrapper : `_ThreadLocalButlerWrapper`
2002 A wrapper that provides access to a thread-local QBB, constructing]
2003 it on first use.
2004 """
2005 dataset_ids = []
2006 for task_label in qg.pipeline_graph.tasks.keys():
2007 for quantum in qg.get_task_quanta(task_label).values():
2008 dataset_ids.append(quantum.outputs[LOG_OUTPUT_TEMPLATE.format(label=task_label)][0].id)
2009 dataset_ids.append(quantum.outputs[METADATA_OUTPUT_TEMPLATE.format(label=task_label)][0].id)
2010 try:
2011 butler_config = full_butler._config # type: ignore[attr-defined]
2012 except AttributeError:
2013 raise RuntimeError("use_qbb=True requires a direct butler.") from None
2014 factory = _QuantumBackedButlerFactory(
2015 butler_config,
2016 dataset_ids,
2017 full_butler.dimensions,
2018 dataset_types={dt.name: dt for dt in qg.registryDatasetTypes()},
2019 )
2020 return cls(factory)
2022 @classmethod
2023 def wrap_full(cls, full_butler: Butler) -> _ThreadLocalButlerWrapper:
2024 """Wrap a full `~lsst.daf.butler.Butler`.
2026 Parameters
2027 ----------
2028 full_butler : `~lsst.daf.butler.Butler`
2029 Full butler to clone when making thread-local copies.
2031 Returns
2032 -------
2033 wrapper : `_ThreadLocalButlerWrapper`
2034 A wrapper that provides access to a thread-local butler,
2035 constructing it on first use.
2036 """
2037 return cls(full_butler.clone)
2039 @property
2040 def butler(self) -> LimitedButler:
2041 """The wrapped butler, constructed on first use within each thread."""
2042 if (butler := getattr(self._thread_local, "butler", None)) is None:
2043 self._thread_local.butler = self._factory()
2044 butler = self._thread_local.butler
2045 return butler
2048@dataclasses.dataclass
2049class _QuantumBackedButlerFactory:
2050 """A factory for `~lsst.daf.butler.QuantumBackedButler`, for use by
2051 `_ThreadLocalButlerWrapper`.
2052 """
2054 config: ButlerConfig
2055 dataset_ids: list[DatasetId]
2056 universe: DimensionUniverse
2057 dataset_types: dict[str, DatasetType]
2059 def __call__(self) -> QuantumBackedButler:
2060 return QuantumBackedButler.from_predicted(
2061 self.config,
2062 predicted_inputs=self.dataset_ids,
2063 predicted_outputs=[],
2064 dimensions=self.universe,
2065 # We don't need the datastore records in the QG because we're
2066 # only going to read metadata and logs, and those are never
2067 # overall inputs.
2068 datastore_records={},
2069 dataset_types=self.dataset_types,
2070 )
2073def _cli() -> None:
2074 import argparse
2076 from .pipeline_graph.visualization import (
2077 QuantumProvenanceGraphStatusAnnotator,
2078 QuantumProvenanceGraphStatusOptions,
2079 show,
2080 )
2082 parser = argparse.ArgumentParser(
2083 "QuantumProvenanceGraph command-line utilities.",
2084 description=(
2085 "This is a small, low-effort debugging utility. "
2086 "It may disappear at any time in favor of a public 'pipetask' interface."
2087 ),
2088 )
2089 subparsers = parser.add_subparsers(dest="cmd")
2091 pprint_parser = subparsers.add_parser("pprint", help="Print a saved summary as a series of tables.")
2092 pprint_parser.add_argument("file", type=argparse.FileType("r"), help="Saved summary JSON file.")
2093 pprint_parser.add_argument(
2094 "--brief",
2095 action=argparse.BooleanOptionalAction,
2096 default=False,
2097 help="Whether to print per-data ID information.",
2098 )
2099 pprint_parser.add_argument(
2100 "--datasets",
2101 action=argparse.BooleanOptionalAction,
2102 default=True,
2103 )
2105 xgraph_parser = subparsers.add_parser("xgraph", help="Print a visual representation of a saved xgraph.")
2106 xgraph_parser.add_argument("file", type=argparse.FileType("r"), help="Saved summary JSON file.")
2107 xgraph_parser.add_argument("qgraph", type=str, help="Saved quantum graph file.")
2109 args = parser.parse_args()
2111 match args.cmd:
2112 case "pprint":
2113 summary = Summary.model_validate_json(args.file.read())
2114 args.file.close()
2115 summary.pprint(brief=args.brief, datasets=args.datasets)
2116 case "xgraph":
2117 summary = Summary.model_validate_json(args.file.read())
2118 args.file.close()
2119 status_annotator = QuantumProvenanceGraphStatusAnnotator(summary)
2120 status_options = QuantumProvenanceGraphStatusOptions(
2121 display_percent=True, display_counts=True, abbreviate=True, visualize=True
2122 )
2123 qgraph = QuantumGraph.loadUri(args.qgraph)
2124 pipeline_graph = qgraph.pipeline_graph
2125 show(
2126 pipeline_graph,
2127 dataset_types=True,
2128 status_annotator=status_annotator,
2129 status_options=status_options,
2130 )
2131 case _:
2132 raise AssertionError(f"Unhandled subcommand {args.dest}.")
2135if __name__ == "__main__":
2136 _cli()