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