Coverage for python / lsst / pipe / base / _status.py: 54%
124 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-27 01:20 -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/>.
28from __future__ import annotations
30__all__ = (
31 "AlgorithmError",
32 "AnnotatedPartialOutputsError",
33 "ExceptionInfo",
34 "InvalidQuantumError",
35 "NoWorkFound",
36 "QuantumAttemptStatus",
37 "QuantumSuccessCaveats",
38 "RepeatableQuantumError",
39 "UnprocessableDataError",
40 "UpstreamFailureNoWorkFound",
41)
43import abc
44import enum
45import logging
46import sys
47from typing import TYPE_CHECKING, Any, ClassVar, Protocol
49import pydantic
51from lsst.utils import introspection
52from lsst.utils.logging import LsstLogAdapter, getLogger
54from ._task_metadata import GetSetDictMetadata, NestedMetadataDict
56if TYPE_CHECKING:
57 from ._task_metadata import TaskMetadata
60_LOG = getLogger(__name__)
63class QuantumSuccessCaveats(enum.Flag):
64 """Flags that add caveats to a "successful" quantum.
66 Quanta can be considered successful even if they do not produce some of
67 their expected outputs (and even if they do not produce all of their
68 expected outputs), as long as the condition is sufficiently well understood
69 that downstream processing should succeed.
70 """
72 NO_CAVEATS = 0
73 """All outputs were produced and no exceptions were raised."""
75 ANY_OUTPUTS_MISSING = enum.auto()
76 """At least one predicted output was not produced."""
78 ALL_OUTPUTS_MISSING = enum.auto()
79 """No predicted outputs (except logs and metadata) were produced.
81 `ANY_OUTPUTS_MISSING` is also set whenever this flag is set.
82 """
84 NO_WORK = enum.auto()
85 """A subclass of `NoWorkFound` was raised.
87 This does not necessarily imply that `ANY_OUTPUTS_MISSING` is not set,
88 since a `PipelineTask.runQuantum` implementation could raise it after
89 directly writing all of its predicted outputs.
90 """
92 ADJUST_QUANTUM_RAISED = enum.auto()
93 """`NoWorkFound` was raised by `PipelineTaskConnnections.adjustQuantum`.
95 This indicates that if a new `QuantumGraph` had been generated immediately
96 before running this quantum, that quantum would not have even been
97 included, because required inputs that were expected to exist by the time
98 it was run (in the original `QuantumGraph`) were not actually produced.
100 `NO_WORK` and `ALL_OUTPUTS_MISSING` are also set whenever this flag is set.
101 """
103 UPSTREAM_FAILURE_NO_WORK = enum.auto()
104 """`UpstreamFailureNoWorkFound` was raised by `PipelineTask.runQuantum`.
106 This exception is raised by downstream tasks when an upstream task's
107 outputs were incomplete in a way that blocks it from running, often
108 because the upstream task raised `AnnotatedPartialOutputsError`.
110 `NO_WORK` is also set whenever this flag is set.
111 """
113 UNPROCESSABLE_DATA = enum.auto()
114 """`UnprocessableDataError` was raised by `PipelineTask.runQuantum`.
116 `NO_WORK` is also set whenever this flag is set.
117 """
119 PARTIAL_OUTPUTS_ERROR = enum.auto()
120 """`AnnotatedPartialOutputsError` was raised by `PipelineTask.runQuantum`
121 and the execution system was instructed to consider this a qualified
122 success.
123 """
125 @classmethod
126 def from_adjust_quantum_no_work(cls) -> QuantumSuccessCaveats:
127 """Return the set of flags appropriate for a quantum for which
128 `PipelineTaskConnections.adjustdQuantum` raised `NoWorkFound`.
129 """
130 return cls.NO_WORK | cls.ADJUST_QUANTUM_RAISED | cls.ANY_OUTPUTS_MISSING | cls.ALL_OUTPUTS_MISSING
132 def concise(self) -> str:
133 """Return a concise string representation of the flags.
135 Returns
136 -------
137 s : `str`
138 Two-character string representation, with the first character
139 indicating whether any predicted outputs were missing and the
140 second representing any exceptions raised. This representation is
141 not always complete; some rare combinations of flags are displayed
142 as if only one of the flags was set.
144 Notes
145 -----
146 The `legend` method returns a description of the returned codes.
147 """
148 char1 = ""
149 if self & QuantumSuccessCaveats.ALL_OUTPUTS_MISSING:
150 char1 = "*"
151 elif self & QuantumSuccessCaveats.ANY_OUTPUTS_MISSING:
152 char1 = "+"
153 char2 = ""
154 if self & QuantumSuccessCaveats.ADJUST_QUANTUM_RAISED:
155 char2 = "A"
156 elif self & QuantumSuccessCaveats.UNPROCESSABLE_DATA:
157 char2 = "D"
158 elif self & QuantumSuccessCaveats.UPSTREAM_FAILURE_NO_WORK:
159 char2 = "U"
160 elif self & QuantumSuccessCaveats.PARTIAL_OUTPUTS_ERROR:
161 char2 = "P"
162 elif self & QuantumSuccessCaveats.NO_WORK:
163 char2 = "N"
164 return char1 + char2
166 @staticmethod
167 def legend() -> dict[str, str]:
168 """Return a `dict` with human-readable descriptions of the characters
169 used in `concise`.
171 Returns
172 -------
173 legend : `dict` [ `str`, `str` ]
174 Mapping from character code to description.
175 """
176 return {
177 "+": "at least one predicted output was missing, but not all were",
178 "*": "all predicted outputs were missing (besides logs and metadata)",
179 "A": "adjustQuantum raised NoWorkFound; a regenerated QG would not include this quantum",
180 "D": "algorithm considers data too bad to be processable",
181 "U": "one or more input dataset was incomplete due to an upstream failure",
182 "P": "task failed but wrote partial outputs; considered a partial success",
183 "N": "runQuantum raised NoWorkFound",
184 }
187class ExceptionInfo(pydantic.BaseModel):
188 """Information about an exception that was raised."""
190 type_name: str
191 """Fully-qualified Python type name for the exception raised."""
193 message: str
194 """String message included in the exception."""
196 metadata: dict[str, float | int | str | bool | None]
197 """Additional metadata included in the exception."""
199 @classmethod
200 def _from_metadata(cls, md: TaskMetadata) -> ExceptionInfo:
201 """Construct from task metadata.
203 Parameters
204 ----------
205 md : `TaskMetadata`
206 Metadata about the error, as written by
207 `AnnotatedPartialOutputsError`.
209 Returns
210 -------
211 info : `ExceptionInfo`
212 Information about the exception.
213 """
214 result = cls(type_name=md["type"], message=md["message"], metadata={})
215 if "metadata" in md:
216 raw_err_metadata = md["metadata"].to_dict()
217 for k, v in raw_err_metadata.items():
218 # Guard against error metadata we wouldn't be able to serialize
219 # later via Pydantic; don't want one weird value bringing down
220 # our ability to report on an entire run.
221 if isinstance(v, float | int | str | bool):
222 result.metadata[k] = v
223 else:
224 _LOG.debug(
225 "Not propagating nested or JSON-incompatible exception metadata key %s=%r.", k, v
226 )
227 return result
229 # Work around the fact that Sphinx chokes on Pydantic docstring formatting,
230 # when we inherit those docstrings in our public classes.
231 if "sphinx" in sys.modules and not TYPE_CHECKING:
233 def copy(self, *args: Any, **kwargs: Any) -> Any:
234 """See `pydantic.BaseModel.copy`."""
235 return super().copy(*args, **kwargs)
237 def model_dump(self, *args: Any, **kwargs: Any) -> Any:
238 """See `pydantic.BaseModel.model_dump`."""
239 return super().model_dump(*args, **kwargs)
241 def model_dump_json(self, *args: Any, **kwargs: Any) -> Any:
242 """See `pydantic.BaseModel.model_dump_json`."""
243 return super().model_dump(*args, **kwargs)
245 def model_copy(self, *args: Any, **kwargs: Any) -> Any:
246 """See `pydantic.BaseModel.model_copy`."""
247 return super().model_copy(*args, **kwargs)
249 @classmethod
250 def model_construct(cls, *args: Any, **kwargs: Any) -> Any: # type: ignore[misc, override]
251 """See `pydantic.BaseModel.model_construct`."""
252 return super().model_construct(*args, **kwargs)
254 @classmethod
255 def model_json_schema(cls, *args: Any, **kwargs: Any) -> Any:
256 """See `pydantic.BaseModel.model_json_schema`."""
257 return super().model_json_schema(*args, **kwargs)
259 @classmethod
260 def model_validate(cls, *args: Any, **kwargs: Any) -> Any:
261 """See `pydantic.BaseModel.model_validate`."""
262 return super().model_validate(*args, **kwargs)
264 @classmethod
265 def model_validate_json(cls, *args: Any, **kwargs: Any) -> Any:
266 """See `pydantic.BaseModel.model_validate_json`."""
267 return super().model_validate_json(*args, **kwargs)
269 @classmethod
270 def model_validate_strings(cls, *args: Any, **kwargs: Any) -> Any:
271 """See `pydantic.BaseModel.model_validate_strings`."""
272 return super().model_validate_strings(*args, **kwargs)
275class QuantumAttemptStatus(enum.Enum):
276 """Enum summarizing an attempt to run a quantum."""
278 ABORTED = -4
279 """The quantum failed with a hard error that prevented both logs and
280 metadata from being written.
282 This state is only set if information from higher-level tooling (e.g. BPS)
283 is available to distinguish it from ``UNKNOWN``.
284 """
286 UNKNOWN = -3
287 """The status of this attempt is unknown.
289 This means no logs or metadata were written, and it at least could not be
290 determined whether the quantum was blocked by an upstream failure (if it
291 was definitely blocked, `BLOCKED` is set instead).
292 """
294 ABORTED_SUCCESS = -2
295 """Task metadata was written for this attempt but logs were not.
297 This is a rare condition that requires a hard failure (i.e. the kind that
298 can prevent a ``finally`` block from running or I/O from being durable) at
299 a very precise time.
300 """
302 FAILED = -1
303 """Execution of the quantum failed gracefully.
305 This is always set if the task metadata dataset was not written but logs
306 were, as is the case when a Python exception is caught and handled by the
307 execution system.
309 This status guarantees that the task log dataset was produced but the
310 metadata dataset was not.
311 """
313 BLOCKED = 0
314 """This quantum was not executed because an upstream quantum failed.
316 Upstream quanta with status `UNKNOWN`, `FAILED`, or `ABORTED` are
317 considered blockers; `ABORTED_SUCCESS` is not.
318 """
320 SUCCESSFUL = 1
321 """This quantum was successfully executed.
323 Quanta may be considered successful even if they do not write any outputs
324 or shortcut early by raising `NoWorkFound` or one of its variants. They
325 may even be considered successful if they raise
326 `AnnotatedPartialOutputsError` if the executor is configured to treat that
327 exception as a non-failure. See `QuantumSuccessCaveats` for details on how
328 these "successes with caveats" are reported.
329 """
331 @property
332 def has_metadata(self) -> bool:
333 """Whether the task metadata dataset was produced."""
334 return self is self.SUCCESSFUL or self is self.ABORTED_SUCCESS
336 @property
337 def has_log(self) -> bool:
338 """Whether the log dataset was produced."""
339 return self is self.SUCCESSFUL or self is self.FAILED
342class GetSetDictMetadataHolder(Protocol):
343 """Protocol for objects that have a ``metadata`` attribute that satisfies
344 `GetSetDictMetadata`.
345 """
347 @property
348 def metadata(self) -> GetSetDictMetadata | None:
349 pass
352class NoWorkFound(BaseException):
353 """An exception raised when a Quantum should not exist because there is no
354 work for it to do.
356 This usually occurs because a non-optional input dataset is not present, or
357 a spatiotemporal overlap that was conservatively predicted does not
358 actually exist.
360 This inherits from BaseException because it is used to signal a case that
361 we don't consider a real error, even though we often want to use try/except
362 logic to trap it.
363 """
365 FLAGS: ClassVar = QuantumSuccessCaveats.NO_WORK
368class UpstreamFailureNoWorkFound(NoWorkFound):
369 """A specialization of `NoWorkFound` that indicates that an upstream task
370 had a problem that was ignored (e.g. to prevent a single-detector failure
371 from bringing down an entire visit).
372 """
374 FLAGS: ClassVar = QuantumSuccessCaveats.NO_WORK | QuantumSuccessCaveats.UPSTREAM_FAILURE_NO_WORK
377class RepeatableQuantumError(RuntimeError):
378 """Exception that may be raised by PipelineTasks (and code they delegate
379 to) in order to indicate that a repeatable problem that will not be
380 addressed by retries.
382 This usually indicates that the algorithm and the data it has been given
383 are somehow incompatible, and the task should run fine on most other data.
385 This exception may be used as a base class for more specific questions, or
386 used directly while chaining another exception, e.g.::
388 try:
389 run_code()
390 except SomeOtherError as err:
391 raise RepeatableQuantumError() from err
393 This may be used for missing input data when the desired behavior is to
394 cause all downstream tasks being run be blocked, forcing the user to
395 address the problem. When the desired behavior is to skip all of this
396 quantum and attempt downstream tasks (or skip them) without its its
397 outputs, raise `NoWorkFound` or return without raising instead.
398 """
400 EXIT_CODE = 20
403class AlgorithmError(RepeatableQuantumError, abc.ABC):
404 """Exception that may be raised by PipelineTasks (and code they delegate
405 to) in order to indicate a repeatable algorithmic failure that will not be
406 addressed by retries.
408 Subclass this exception to define the metadata associated with the error
409 (for example: number of data points in a fit vs. degrees of freedom).
410 """
412 def __new__(cls, *args: Any, **kwargs: Any) -> AlgorithmError:
413 # Have to override __new__ because builtin subclasses aren't checked
414 # for abstract methods; see https://github.com/python/cpython/issues/50246
415 if cls.__abstractmethods__:
416 raise TypeError(
417 f"Can't instantiate abstract class {cls.__name__} with "
418 f"abstract methods: {','.join(sorted(cls.__abstractmethods__))}"
419 )
420 return super().__new__(cls, *args, **kwargs)
422 @property
423 @abc.abstractmethod
424 def metadata(self) -> NestedMetadataDict | None:
425 """Metadata from the raising `~lsst.pipe.base.Task` with more
426 information about the failure. The contents of the dict are
427 `~lsst.pipe.base.Task`-dependent, and must have `str` keys and `str`,
428 `int`, `float`, `bool`, or nested-dictionary (with the same key and
429 value types) values.
430 """
431 raise NotImplementedError
434class UnprocessableDataError(NoWorkFound):
435 """A specialization of `NoWorkFound` that will be [subclassed and] raised
436 by Tasks to indicate a failure to process their inputs for some reason that
437 is non-recoverable.
439 Notes
440 -----
441 An example is a known bright star that causes PSF measurement to fail, and
442 that makes that detector entirely non-recoverable. Another example is an
443 image with an oddly shaped PSF (e.g. due to a failure to achieve focus)
444 that warrants the image being flagged as "poor quality" which should not
445 have further processing attempted.
447 The `NoWorkFound` inheritance ensures the job will not be considered a
448 failure (i.e. such that no human time will inadvertently be spent chasing
449 it down).
451 Do not raise this unless we are convinced that the data cannot (or should
452 not) be processed, even by a better algorithm. Most instances where this
453 error would be raised likely require an RFC to explicitly define the
454 situation.
455 """
457 FLAGS: ClassVar = QuantumSuccessCaveats.NO_WORK | QuantumSuccessCaveats.UNPROCESSABLE_DATA
460class AnnotatedPartialOutputsError(RepeatableQuantumError):
461 """Exception that runQuantum raises when the (partial) outputs it has
462 written contain information about their own incompleteness or degraded
463 quality.
465 Clients should construct this exception by calling `annotate` instead of
466 calling the constructor directly. However, `annotate` does not chain the
467 exception; this must still be done by the client.
469 This exception should always chain the original error. When the
470 executor catches this exception, it will report the original exception. In
471 contrast, other exceptions raised from ``runQuantum`` are considered to
472 invalidate any outputs that are already written.
473 """
475 FLAGS: ClassVar = QuantumSuccessCaveats.PARTIAL_OUTPUTS_ERROR
477 @classmethod
478 def annotate(
479 cls, error: Exception, *args: GetSetDictMetadataHolder | None, log: logging.Logger | LsstLogAdapter
480 ) -> AnnotatedPartialOutputsError:
481 """Set metadata on outputs to explain the nature of the failure.
483 Parameters
484 ----------
485 error : `Exception`
486 Exception that caused the task to fail.
487 *args : `GetSetDictMetadataHolder`
488 Objects (e.g. Task, Exposure, SimpleCatalog) to annotate with
489 failure information. They must have a `metadata` property.
490 log : `logging.Logger`
491 Log to send error message to.
493 Returns
494 -------
495 error : `AnnotatedPartialOutputsError`
496 Exception that the failing task can ``raise from`` with the
497 passed-in exception.
499 Notes
500 -----
501 This should be called from within an except block that has caught an
502 exception. Here is an example of handling a failure in
503 ``PipelineTask.runQuantum`` that annotates and writes partial outputs:
505 .. code-block:: py
506 :name: annotate-error-example
508 def runQuantum(self, butlerQC, inputRefs, outputRefs):
509 inputs = butlerQC.get(inputRefs)
510 exposures = inputs.pop("exposures")
511 assert not inputs, "runQuantum got more inputs than expected"
513 result = pipeBase.Struct(catalog=None)
514 try:
515 self.run(exposure)
516 except pipeBase.AlgorithmError as e:
517 error = pipeBase.AnnotatedPartialOutputsError.annotate(
518 e, self, result.catalog, log=self.log
519 )
520 raise error from e
521 finally:
522 butlerQC.put(result, outputRefs)
523 """
524 failure_info = {
525 "message": str(error),
526 "type": introspection.get_full_type_name(error),
527 }
528 if other := getattr(error, "metadata", None):
529 failure_info["metadata"] = other
531 # NOTE: Can't fully test this in pipe_base because afw is not a
532 # dependency; test_calibrateImage.py in pipe_tasks gives more coverage.
533 for item in args:
534 # Some outputs may not exist, so we cannot set metadata on them.
535 if item is None:
536 continue
537 item.metadata.set_dict("failure", failure_info) # type: ignore
539 log.debug(
540 "Task failed with only partial outputs; see exception message for details.",
541 exc_info=error,
542 )
544 return cls("Task failed and wrote partial outputs: see chained exception for details.")
547class InvalidQuantumError(Exception):
548 """Exception that may be raised by PipelineTasks (and code they delegate
549 to) in order to indicate logic bug or configuration problem.
551 This usually indicates that the configured algorithm itself is invalid and
552 will not run on a significant fraction of quanta (often all of them).
554 This exception may be used as a base class for more specific questions, or
555 used directly while chaining another exception, e.g.::
557 try:
558 run_code()
559 except SomeOtherError as err:
560 raise RepeatableQuantumError() from err
562 Raising this exception in `PipelineTask.runQuantum` or something it calls
563 is a last resort - whenever possible, such problems should cause exceptions
564 in ``__init__`` or in QuantumGraph generation. It should never be used
565 for missing data.
566 """
568 EXIT_CODE = 21