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