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