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/>. 

27 

28from __future__ import annotations 

29 

30import abc 

31import enum 

32import logging 

33from typing import TYPE_CHECKING, Any, ClassVar, Protocol 

34 

35from lsst.utils import introspection 

36 

37from ._task_metadata import GetSetDictMetadata, NestedMetadataDict 

38 

39if TYPE_CHECKING: 

40 from lsst.utils.logging import LsstLogAdapter 

41 

42__all__ = ( 

43 "AlgorithmError", 

44 "AnnotatedPartialOutputsError", 

45 "InvalidQuantumError", 

46 "NoWorkFound", 

47 "QuantumSuccessCaveats", 

48 "RepeatableQuantumError", 

49 "UnprocessableDataError", 

50 "UpstreamFailureNoWorkFound", 

51) 

52 

53 

54class QuantumSuccessCaveats(enum.Flag): 

55 """Flags that add caveats to a "successful" quantum. 

56 

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 """ 

62 

63 NO_CAVEATS = 0 

64 """All outputs were produced and no exceptions were raised.""" 

65 

66 ANY_OUTPUTS_MISSING = enum.auto() 

67 """At least one predicted output was not produced.""" 

68 

69 ALL_OUTPUTS_MISSING = enum.auto() 

70 """No predicted outputs (except logs and metadata) were produced. 

71 

72 `ANY_OUTPUTS_MISSING` is also set whenever this flag is set. 

73 """ 

74 

75 NO_WORK = enum.auto() 

76 """A subclass of `NoWorkFound` was raised. 

77 

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 """ 

82 

83 ADJUST_QUANTUM_RAISED = enum.auto() 

84 """`NoWorkFound` was raised by `PipelineTaskConnnections.adjustQuantum`. 

85 

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. 

90 

91 `NO_WORK` and `ALL_OUTPUTS_MISSING` are also set whenever this flag is set. 

92 """ 

93 

94 UPSTREAM_FAILURE_NO_WORK = enum.auto() 

95 """`UpstreamFailureNoWorkFound` was raised by `PipelineTask.runQuantum`. 

96 

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`. 

100 

101 `NO_WORK` is also set whenever this flag is set. 

102 """ 

103 

104 UNPROCESSABLE_DATA = enum.auto() 

105 """`UnprocessableDataError` was raised by `PipelineTask.runQuantum`. 

106 

107 `NO_WORK` is also set whenever this flag is set. 

108 """ 

109 

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 """ 

115 

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 

122 

123 def concise(self) -> str: 

124 """Return a concise string representation of the flags. 

125 

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. 

134 

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 

156 

157 @staticmethod 

158 def legend() -> dict[str, str]: 

159 """Return a `dict` with human-readable descriptions of the characters 

160 used in `concise`. 

161 

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 } 

176 

177 

178class GetSetDictMetadataHolder(Protocol): 

179 """Protocol for objects that have a ``metadata`` attribute that satisfies 

180 `GetSetDictMetadata`. 

181 """ 

182 

183 @property 

184 def metadata(self) -> GetSetDictMetadata | None: 

185 pass 

186 

187 

188class NoWorkFound(BaseException): 

189 """An exception raised when a Quantum should not exist because there is no 

190 work for it to do. 

191 

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. 

195 

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 """ 

200 

201 FLAGS: ClassVar = QuantumSuccessCaveats.NO_WORK 

202 

203 

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 """ 

209 

210 FLAGS: ClassVar = QuantumSuccessCaveats.NO_WORK | QuantumSuccessCaveats.UPSTREAM_FAILURE_NO_WORK 

211 

212 

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. 

217 

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. 

220 

221 This exception may be used as a base class for more specific questions, or 

222 used directly while chaining another exception, e.g.:: 

223 

224 try: 

225 run_code() 

226 except SomeOtherError as err: 

227 raise RepeatableQuantumError() from err 

228 

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 """ 

235 

236 EXIT_CODE = 20 

237 

238 

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. 

243 

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 """ 

247 

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) 

257 

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 

268 

269 

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. 

274 

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. 

282 

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). 

286 

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 """ 

292 

293 FLAGS: ClassVar = QuantumSuccessCaveats.NO_WORK | QuantumSuccessCaveats.UNPROCESSABLE_DATA 

294 

295 

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. 

300 

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. 

304 

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 """ 

310 

311 FLAGS: ClassVar = QuantumSuccessCaveats.PARTIAL_OUTPUTS_ERROR 

312 

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. 

318 

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. 

328 

329 Returns 

330 ------- 

331 error : `AnnotatedPartialOutputsError` 

332 Exception that the failing task can ``raise from`` with the 

333 passed-in exception. 

334 

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: 

340 

341 .. code-block:: py 

342 :name: annotate-error-example 

343 

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" 

348 

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 

366 

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 

374 

375 log.debug( 

376 "Task failed with only partial outputs; see exception message for details.", 

377 exc_info=error, 

378 ) 

379 

380 return cls("Task failed and wrote partial outputs: see chained exception for details.") 

381 

382 

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. 

386 

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). 

389 

390 This exception may be used as a base class for more specific questions, or 

391 used directly while chaining another exception, e.g.:: 

392 

393 try: 

394 run_code() 

395 except SomeOtherError as err: 

396 raise RepeatableQuantumError() from err 

397 

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 """ 

403 

404 EXIT_CODE = 21