Coverage for python/lsst/pipe/base/_status.py: 49%
84 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-09 03:38 -0700
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/>.
28from __future__ import annotations
30import abc
31import enum
32import logging
33from typing import TYPE_CHECKING, Any, ClassVar, Protocol
35from lsst.utils import introspection
37from ._task_metadata import GetSetDictMetadata, NestedMetadataDict
39if TYPE_CHECKING:
40 from lsst.utils.logging import LsstLogAdapter
42__all__ = (
43 "AlgorithmError",
44 "AnnotatedPartialOutputsError",
45 "InvalidQuantumError",
46 "NoWorkFound",
47 "QuantumSuccessCaveats",
48 "RepeatableQuantumError",
49 "UnprocessableDataError",
50 "UpstreamFailureNoWorkFound",
51)
54class QuantumSuccessCaveats(enum.Flag):
55 """Flags that add caveats to a "successful" quantum.
57 Quanta can be considered successful even if they do not produce some of
58 their expected outputs (and even if they do not produce all of their
59 expected outputs), as long as the condition is sufficiently well understood
60 that downstream processing should succeed.
61 """
63 NO_CAVEATS = 0
64 """All outputs were produced and no exceptions were raised."""
66 ANY_OUTPUTS_MISSING = enum.auto()
67 """At least one predicted output was not produced."""
69 ALL_OUTPUTS_MISSING = enum.auto()
70 """No predicted outputs (except logs and metadata) were produced.
72 `ANY_OUTPUTS_MISSING` is also set whenever this flag is set.
73 """
75 NO_WORK = enum.auto()
76 """A subclass of `NoWorkFound` was raised.
78 This does not necessarily imply that `ANY_OUTPUTS_MISSING` is not set,
79 since a `PipelineTask.runQuantum` implementation could raise it after
80 directly writing all of its predicted outputs.
81 """
83 ADJUST_QUANTUM_RAISED = enum.auto()
84 """`NoWorkFound` was raised by `PipelineTaskConnnections.adjustQuantum`.
86 This indicates that if a new `QuantumGraph` had been generated immediately
87 before running this quantum, that quantum would not have even been
88 included, because required inputs that were expected to exist by the time
89 it was run (in the original `QuantumGraph`) were not actually produced.
91 `NO_WORK` and `ALL_OUTPUTS_MISSING` are also set whenever this flag is set.
92 """
94 UPSTREAM_FAILURE_NO_WORK = enum.auto()
95 """`UpstreamFailureNoWorkFound` was raised by `PipelineTask.runQuantum`.
97 This exception is raised by downstream tasks when an upstream task's
98 outputs were incomplete in a way that blocks it from running, often
99 because the upstream task raised `AnnotatedPartialOutputsError`.
101 `NO_WORK` is also set whenever this flag is set.
102 """
104 UNPROCESSABLE_DATA = enum.auto()
105 """`UnprocessableDataError` was raised by `PipelineTask.runQuantum`.
107 `NO_WORK` is also set whenever this flag is set.
108 """
110 PARTIAL_OUTPUTS_ERROR = enum.auto()
111 """`AnnotatedPartialOutputsError` was raised by `PipelineTask.runQuantum`
112 and the execution system was instructed to consider this a qualified
113 success.
114 """
116 @classmethod
117 def from_adjust_quantum_no_work(cls) -> QuantumSuccessCaveats:
118 """Return the set of flags appropriate for a quantum for which
119 `PipelineTaskConnections.adjustdQuantum` raised `NoWorkFound`.
120 """
121 return cls.NO_WORK | cls.ADJUST_QUANTUM_RAISED | cls.ANY_OUTPUTS_MISSING | cls.ALL_OUTPUTS_MISSING
123 def concise(self) -> str:
124 """Return a concise string representation of the flags.
126 Returns
127 -------
128 s : `str`
129 Two-character string representation, with the first character
130 indicating whether any predicted outputs were missing and the
131 second representing any exceptions raised. This representation is
132 not always complete; some rare combinations of flags are displayed
133 as if only one of the flags was set.
135 Notes
136 -----
137 The `legend` method returns a description of the returned codes.
138 """
139 char1 = ""
140 if self & QuantumSuccessCaveats.ALL_OUTPUTS_MISSING:
141 char1 = "*"
142 elif self & QuantumSuccessCaveats.ANY_OUTPUTS_MISSING:
143 char1 = "+"
144 char2 = ""
145 if self & QuantumSuccessCaveats.ADJUST_QUANTUM_RAISED:
146 char2 = "A"
147 elif self & QuantumSuccessCaveats.UNPROCESSABLE_DATA:
148 char2 = "D"
149 elif self & QuantumSuccessCaveats.UPSTREAM_FAILURE_NO_WORK:
150 char2 = "U"
151 elif self & QuantumSuccessCaveats.PARTIAL_OUTPUTS_ERROR:
152 char2 = "P"
153 elif self & QuantumSuccessCaveats.NO_WORK:
154 char2 = "N"
155 return char1 + char2
157 @staticmethod
158 def legend() -> dict[str, str]:
159 """Return a `dict` with human-readable descriptions of the characters
160 used in `concise`.
162 Returns
163 -------
164 legend : `dict` [ `str`, `str` ]
165 Mapping from character code to description.
166 """
167 return {
168 "+": "at least one predicted output was missing, but not all were",
169 "*": "all predicated outputs were missing (besides logs and metadata)",
170 "A": "adjustQuantum raised NoWorkFound; a regenerated QG would not include this quantum",
171 "D": "algorithm considers data too bad to be processable",
172 "U": "one or more input dataset was incomplete due to an upstream failure",
173 "P": "task failed but wrote partial outputs; considered a partial success",
174 "N": "runQuantum raised NoWorkFound",
175 }
178class GetSetDictMetadataHolder(Protocol):
179 """Protocol for objects that have a ``metadata`` attribute that satisfies
180 `GetSetDictMetadata`.
181 """
183 @property
184 def metadata(self) -> GetSetDictMetadata | None:
185 pass
188class NoWorkFound(BaseException):
189 """An exception raised when a Quantum should not exist because there is no
190 work for it to do.
192 This usually occurs because a non-optional input dataset is not present, or
193 a spatiotemporal overlap that was conservatively predicted does not
194 actually exist.
196 This inherits from BaseException because it is used to signal a case that
197 we don't consider a real error, even though we often want to use try/except
198 logic to trap it.
199 """
201 FLAGS: ClassVar = QuantumSuccessCaveats.NO_WORK
204class UpstreamFailureNoWorkFound(NoWorkFound):
205 """A specialization of `NoWorkFound` that indicates that an upstream task
206 had a problem that was ignored (e.g. to prevent a single-detector failure
207 from bringing down an entire visit).
208 """
210 FLAGS: ClassVar = QuantumSuccessCaveats.NO_WORK | QuantumSuccessCaveats.UPSTREAM_FAILURE_NO_WORK
213class RepeatableQuantumError(RuntimeError):
214 """Exception that may be raised by PipelineTasks (and code they delegate
215 to) in order to indicate that a repeatable problem that will not be
216 addressed by retries.
218 This usually indicates that the algorithm and the data it has been given
219 are somehow incompatible, and the task should run fine on most other data.
221 This exception may be used as a base class for more specific questions, or
222 used directly while chaining another exception, e.g.::
224 try:
225 run_code()
226 except SomeOtherError as err:
227 raise RepeatableQuantumError() from err
229 This may be used for missing input data when the desired behavior is to
230 cause all downstream tasks being run be blocked, forcing the user to
231 address the problem. When the desired behavior is to skip all of this
232 quantum and attempt downstream tasks (or skip them) without its its
233 outputs, raise `NoWorkFound` or return without raising instead.
234 """
236 EXIT_CODE = 20
239class AlgorithmError(RepeatableQuantumError, abc.ABC):
240 """Exception that may be raised by PipelineTasks (and code they delegate
241 to) in order to indicate a repeatable algorithmic failure that will not be
242 addressed by retries.
244 Subclass this exception to define the metadata associated with the error
245 (for example: number of data points in a fit vs. degrees of freedom).
246 """
248 def __new__(cls, *args: Any, **kwargs: Any) -> AlgorithmError:
249 # Have to override __new__ because builtin subclasses aren't checked
250 # for abstract methods; see https://github.com/python/cpython/issues/50246
251 if cls.__abstractmethods__:
252 raise TypeError(
253 f"Can't instantiate abstract class {cls.__name__} with "
254 f"abstract methods: {','.join(sorted(cls.__abstractmethods__))}"
255 )
256 return super().__new__(cls, *args, **kwargs)
258 @property
259 @abc.abstractmethod
260 def metadata(self) -> NestedMetadataDict | None:
261 """Metadata from the raising `~lsst.pipe.base.Task` with more
262 information about the failure. The contents of the dict are
263 `~lsst.pipe.base.Task`-dependent, and must have `str` keys and `str`,
264 `int`, `float`, `bool`, or nested-dictionary (with the same key and
265 value types) values.
266 """
267 raise NotImplementedError
270class UnprocessableDataError(NoWorkFound):
271 """A specialization of `NoWorkFound` that will be [subclassed and] raised
272 by Tasks to indicate a failure to process their inputs for some reason that
273 is non-recoverable.
275 Notes
276 -----
277 An example is a known bright star that causes PSF measurement to fail, and
278 that makes that detector entirely non-recoverable. Another example is an
279 image with an oddly shaped PSF (e.g. due to a failure to achieve focus)
280 that warrants the image being flagged as "poor quality" which should not
281 have further processing attempted.
283 The `NoWorkFound` inheritance ensures the job will not be considered a
284 failure (i.e. such that no human time will inadvertently be spent chasing
285 it down).
287 Do not raise this unless we are convinced that the data cannot (or should
288 not) be processed, even by a better algorithm. Most instances where this
289 error would be raised likely require an RFC to explicitly define the
290 situation.
291 """
293 FLAGS: ClassVar = QuantumSuccessCaveats.NO_WORK | QuantumSuccessCaveats.UNPROCESSABLE_DATA
296class AnnotatedPartialOutputsError(RepeatableQuantumError):
297 """Exception that runQuantum raises when the (partial) outputs it has
298 written contain information about their own incompleteness or degraded
299 quality.
301 Clients should construct this exception by calling `annotate` instead of
302 calling the constructor directly. However, `annotate` does not chain the
303 exception; this must still be done by the client.
305 This exception should always chain the original error. When the
306 executor catches this exception, it will report the original exception. In
307 contrast, other exceptions raised from ``runQuantum`` are considered to
308 invalidate any outputs that are already written.
309 """
311 FLAGS: ClassVar = QuantumSuccessCaveats.PARTIAL_OUTPUTS_ERROR
313 @classmethod
314 def annotate(
315 cls, error: Exception, *args: GetSetDictMetadataHolder | None, log: logging.Logger | LsstLogAdapter
316 ) -> AnnotatedPartialOutputsError:
317 """Set metadata on outputs to explain the nature of the failure.
319 Parameters
320 ----------
321 error : `Exception`
322 Exception that caused the task to fail.
323 *args : `GetSetDictMetadataHolder`
324 Objects (e.g. Task, Exposure, SimpleCatalog) to annotate with
325 failure information. They must have a `metadata` property.
326 log : `logging.Logger`
327 Log to send error message to.
329 Returns
330 -------
331 error : `AnnotatedPartialOutputsError`
332 Exception that the failing task can ``raise from`` with the
333 passed-in exception.
335 Notes
336 -----
337 This should be called from within an except block that has caught an
338 exception. Here is an example of handling a failure in
339 ``PipelineTask.runQuantum`` that annotates and writes partial outputs:
341 .. code-block:: py
342 :name: annotate-error-example
344 def runQuantum(self, butlerQC, inputRefs, outputRefs):
345 inputs = butlerQC.get(inputRefs)
346 exposures = inputs.pop("exposures")
347 assert not inputs, "runQuantum got more inputs than expected"
349 result = pipeBase.Struct(catalog=None)
350 try:
351 self.run(exposure)
352 except pipeBase.AlgorithmError as e:
353 error = pipeBase.AnnotatedPartialOutputsError.annotate(
354 e, self, result.catalog, log=self.log
355 )
356 raise error from e
357 finally:
358 butlerQC.put(result, outputRefs)
359 """
360 failure_info = {
361 "message": str(error),
362 "type": introspection.get_full_type_name(error),
363 }
364 if other := getattr(error, "metadata", None):
365 failure_info["metadata"] = other
367 # NOTE: Can't fully test this in pipe_base because afw is not a
368 # dependency; test_calibrateImage.py in pipe_tasks gives more coverage.
369 for item in args:
370 # Some outputs may not exist, so we cannot set metadata on them.
371 if item is None:
372 continue
373 item.metadata.set_dict("failure", failure_info) # type: ignore
375 log.debug(
376 "Task failed with only partial outputs; see exception message for details.",
377 exc_info=error,
378 )
380 return cls("Task failed and wrote partial outputs: see chained exception for details.")
383class InvalidQuantumError(Exception):
384 """Exception that may be raised by PipelineTasks (and code they delegate
385 to) in order to indicate logic bug or configuration problem.
387 This usually indicates that the configured algorithm itself is invalid and
388 will not run on a significant fraction of quanta (often all of them).
390 This exception may be used as a base class for more specific questions, or
391 used directly while chaining another exception, e.g.::
393 try:
394 run_code()
395 except SomeOtherError as err:
396 raise RepeatableQuantumError() from err
398 Raising this exception in `PipelineTask.runQuantum` or something it calls
399 is a last resort - whenever possible, such problems should cause exceptions
400 in ``__init__`` or in QuantumGraph generation. It should never be used
401 for missing data.
402 """
404 EXIT_CODE = 21