Coverage for python/lsst/ctrl/mpexec/singleQuantumExecutor.py: 65%
54 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-16 15:01 -0700
1# This file is part of ctrl_mpexec.
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__ = ("SingleQuantumExecutor",)
32import uuid
33from collections.abc import Callable, Mapping
34from typing import TYPE_CHECKING, Any
36from deprecated.sphinx import deprecated
38import lsst.pipe.base.single_quantum_executor
40if TYPE_CHECKING:
41 from lsst.daf.butler import Butler, ButlerMetrics, LimitedButler, Quantum
42 from lsst.pipe.base import ExecutionResources, PipelineTask, QuantumSuccessCaveats, TaskFactory
43 from lsst.pipe.base.log_capture import _ExecutionLogRecordsExtra
44 from lsst.pipe.base.pipeline_graph import TaskNode
47# TODO[DM-51962]: Remove this module.
48@deprecated(
49 "The SingleQuantumExecutor class has moved to lsst.pipe.base.single_quantum_executor. "
50 "This forwarding shim will be removed after v30.",
51 version="v30",
52 category=FutureWarning,
53)
54class SingleQuantumExecutor(lsst.pipe.base.single_quantum_executor.SingleQuantumExecutor):
55 """Executor class which runs one Quantum at a time.
57 Parameters
58 ----------
59 butler : `~lsst.daf.butler.Butler` or `None`
60 Data butler, `None` means that Quantum-backed butler should be used
61 instead.
62 taskFactory : `~lsst.pipe.base.TaskFactory`
63 Instance of a task factory.
64 skipExistingIn : `~typing.Any`
65 Expressions representing the collections to search for existing output
66 datasets. See :ref:`daf_butler_ordered_collection_searches` for allowed
67 types. This class only checks for the presence of butler output run in
68 the list of collections. If the output run is present in the list then
69 the quanta whose complete outputs exist in the output run will be
70 skipped. `None` or empty string/sequence disables skipping.
71 clobberOutputs : `bool`, optional
72 If `True`, then outputs from a quantum that exist in output run
73 collection will be removed prior to executing a quantum. If
74 ``skipExistingIn`` contains output run, then only partial outputs from
75 a quantum will be removed. Only used when ``butler`` is not `None`.
76 enableLsstDebug : `bool`, optional
77 Enable debugging with ``lsstDebug`` facility for a task.
78 limited_butler_factory : `~collections.abc.Callable`, optional
79 A method that creates a `~lsst.daf.butler.LimitedButler` instance for a
80 given Quantum. This parameter must be defined if ``butler`` is `None`.
81 If ``butler`` is not `None` then this parameter is ignored.
82 resources : `~lsst.pipe.base.ExecutionResources`, optional
83 The resources available to this quantum when executing.
84 skipExisting : `bool`, optional
85 If `True`, skip quanta whose metadata datasets are already stored.
86 Unlike ``skipExistingIn``, this works with limited butlers as well as
87 full butlers. Always set to `True` if ``skipExistingIn`` matches
88 ``butler.run``.
89 assumeNoExistingOutputs : `bool`, optional
90 If `True`, assume preexisting outputs are impossible (e.g. because this
91 is known by higher-level code to be a new ``RUN`` collection), and do
92 not look for them. This causes the ``skipExisting`` and
93 ``clobberOutputs`` options to be ignored, but unlike just setting both
94 of those to `False`, it also avoids all dataset existence checks.
95 raise_on_partial_outputs : `bool`, optional
96 If `True` raise exceptions chained by
97 `lsst.pipe.base.AnnotatedPartialOutputsError` immediately, instead of
98 considering the partial result a success and continuing to run
99 downstream tasks.
100 job_metadata : `~collections.abc.Mapping`
101 Mapping with extra metadata to embed within the quantum metadata under
102 the "job" key. This is intended to correspond to information common
103 to all quanta being executed in a single process, such as the time
104 taken to load the quantum graph in a BPS job.
106 Notes
107 -----
108 This is a deprecated backwards-compatibility shim for
109 `lsst.pipe.base.single_quantum_executor.SingleQuantumExecutor`, which has
110 the same functionality with very minor interface changes.
111 """
113 def __init__(
114 self,
115 butler: Butler | None,
116 taskFactory: TaskFactory,
117 skipExistingIn: Any = None,
118 clobberOutputs: bool = False,
119 enableLsstDebug: bool = False,
120 limited_butler_factory: Callable[[Quantum], LimitedButler] | None = None,
121 resources: ExecutionResources | None = None,
122 skipExisting: bool = False,
123 assumeNoExistingOutputs: bool = False,
124 raise_on_partial_outputs: bool = True,
125 job_metadata: Mapping[str, int | str | float] | None = None,
126 ):
127 super().__init__(
128 butler=butler,
129 task_factory=taskFactory,
130 skip_existing_in=skipExistingIn,
131 clobber_outputs=clobberOutputs,
132 enable_lsst_debug=enableLsstDebug,
133 limited_butler_factory=limited_butler_factory,
134 resources=resources,
135 skip_existing=skipExisting,
136 assume_no_existing_outputs=assumeNoExistingOutputs,
137 raise_on_partial_outputs=raise_on_partial_outputs,
138 job_metadata=job_metadata,
139 )
141 def checkExistingOutputs(
142 self,
143 quantum: Quantum,
144 task_node: TaskNode,
145 /,
146 limited_butler: LimitedButler,
147 log_extra: _ExecutionLogRecordsExtra,
148 ) -> bool:
149 return super()._check_existing_outputs(
150 quantum, task_node, limited_butler=limited_butler, log_extra=log_extra
151 )
153 def updatedQuantumInputs(
154 self, quantum: Quantum, task_node: TaskNode, /, limited_butler: LimitedButler
155 ) -> Quantum:
156 return super()._updated_quantum_inputs(quantum, task_node, limited_butler=limited_butler)
158 def runQuantum(
159 self,
160 task: PipelineTask,
161 quantum: Quantum,
162 task_node: TaskNode,
163 /,
164 limited_butler: LimitedButler,
165 quantum_id: uuid.UUID | None = None,
166 ) -> tuple[QuantumSuccessCaveats, list[uuid.UUID], ButlerMetrics]:
167 ids_put: list[uuid.UUID] = []
168 with limited_butler.record_metrics() as butler_metrics:
169 quantum_success_caveats = super()._run_quantum(
170 task,
171 quantum,
172 task_node,
173 limited_butler=limited_butler,
174 quantum_id=quantum_id,
175 ids_put=ids_put,
176 )
177 return quantum_success_caveats, ids_put, butler_metrics
179 def writeMetadata(
180 self, quantum: Quantum, metadata: Any, task_node: TaskNode, /, limited_butler: LimitedButler
181 ) -> None:
182 return super()._write_metadata(quantum, metadata, task_node, limited_butler=limited_butler)
184 def initGlobals(self, quantum: Quantum) -> None:
185 pass
187 @property
188 def butler(self) -> Butler | None:
189 return self._butler
191 @property
192 def taskFactory(self) -> TaskFactory:
193 return self._task_factory
195 @property
196 def clobberOutputs(self) -> bool:
197 return self._clobber_outputs
199 @property
200 def enableLsstDebug(self) -> bool:
201 return self._enable_lsst_debug
203 @property
204 def limited_butler_factory(self) -> Callable[[Quantum], LimitedButler] | None:
205 return self._limited_butler_factory
207 @property
208 def resources(self) -> ExecutionResources | None:
209 return self._resources
211 @property
212 def assumeNoExistingOutputs(self) -> bool:
213 return self._assume_no_existing_outputs
215 @property
216 def raise_on_partial_outputs(self) -> bool:
217 return self._raise_on_partial_outputs
219 @property
220 def job_metadata(self) -> Mapping[str, int | str | float] | None:
221 return self._job_metadata
223 @property
224 def skipExisting(self) -> bool:
225 return self._skip_existing