Coverage for python / lsst / pipe / base / connectionTypes.py: 75%
63 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-20 01:37 -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/>.
28"""Module defining connection types to be used within a
29`.PipelineTaskConnections` class.
30"""
32__all__ = ["BaseConnection", "InitInput", "InitOutput", "Input", "Output", "PrerequisiteInput"]
34import dataclasses
35from collections.abc import Callable, Iterable, Sequence
36from typing import ClassVar
38from lsst.daf.butler import DataCoordinate, DatasetRef, DatasetType, Registry
39from lsst.utils.introspection import find_outside_stacklevel
42@dataclasses.dataclass(frozen=True)
43class BaseConnection:
44 """Base class used for declaring `PipelineTask` connections.
46 Attributes
47 ----------
48 name : `str`
49 The name used to identify the dataset type.
50 storageClass : `str`
51 The storage class used when (un)/persisting the dataset type.
52 multiple : `bool`
53 Indicates if this connection should expect to contain multiple objects
54 of the given dataset type. Tasks with more than one connection with
55 ``multiple=True`` with the same dimensions may want to implement
56 `.PipelineTaskConnections.adjustQuantum` to ensure those datasets are
57 consistent (i.e. zip-iterable) in `PipelineTask.runQuantum()` and
58 notify the execution system as early as possible of outputs that will
59 not be produced because the corresponding input is missing.
60 deprecated : `str`, optional
61 A description of why this connection is deprecated, including the
62 version after which it may be removed.
64 If not `None`, the string is appended to the docstring for this
65 connection and the corresponding config Field.
66 """
68 name: str
69 storageClass: str
70 doc: str = ""
71 multiple: bool = False
72 deprecated: str | None = dataclasses.field(default=None, kw_only=True)
74 _connection_type_set: ClassVar[str]
75 _deprecation_context: str = ""
77 def __post_init__(self):
78 if self.deprecated and not self._deprecation_context: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true
79 info = {}
80 _ = find_outside_stacklevel("lsst.pipe.base", "dataclasses", stack_info=info)
81 object.__setattr__(self, "_deprecation_context", f"{info['filename']}:{info['lineno']}")
83 def __get__(self, inst, klass):
84 """Descriptor access method.
86 This is a method used to turn a connection into a descriptor.
87 When a connection is added to a connection class, it is a class level
88 variable. This method makes accessing this connection, on the
89 instance of the connection class owning this connection, return a
90 result specialized for that instance. In the case of connections
91 this specifically means names specified in a config instance will
92 be visible instead of the default names for the connection, and that
93 removed connections will not be accessible on the instance.
94 """
95 # If inst is None, this is being accessed by the class and not an
96 # instance, return this connection itself
97 if inst is None:
98 return self
99 # Attempt to return the configured connection object from the
100 # connections instance allConnections mapping.
101 try:
102 return inst.allConnections[self.varName]
103 except KeyError:
104 raise AttributeError(
105 f"Connection {self.varName!r} of {klass.__name__} has been removed."
106 ) from None
109@dataclasses.dataclass(frozen=True)
110class DimensionedConnection(BaseConnection):
111 """Class used for declaring PipelineTask connections that includes
112 dimensions.
114 Attributes
115 ----------
116 name : `str`
117 The name used to identify the dataset type.
118 storageClass : `str`
119 The storage class used when (un)/persisting the dataset type.
120 multiple : `bool`
121 Indicates if this connection should expect to contain multiple objects
122 of the given dataset type. Tasks with more than one connection with
123 ``multiple=True`` with the same dimensions may want to implement
124 `.PipelineTaskConnections.adjustQuantum` to ensure those datasets are
125 consistent (i.e. zip-iterable) in `PipelineTask.runQuantum` and notify
126 the execution system as early as possible of outputs that will not be
127 produced because the corresponding input is missing.
128 dimensions : iterable of `str`
129 The `lsst.daf.butler.Butler` `lsst.daf.butler.Registry` dimensions used
130 to identify the dataset type identified by the specified name.
131 isCalibration : `bool`, optional
132 `True` if this dataset type may be included in CALIBRATION-type
133 collections to associate it with a validity range, `False` (default)
134 otherwise.
135 """
137 dimensions: Iterable[str] = ()
138 isCalibration: bool = False
140 def __post_init__(self):
141 super().__post_init__()
142 if isinstance(self.dimensions, str): 142 ↛ 143line 142 didn't jump to line 143 because the condition on line 142 was never true
143 raise TypeError(
144 "Dimensions must be iterable of dimensions, got str, possibly omitted trailing comma"
145 )
146 if not isinstance(self.dimensions, Iterable): 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true
147 raise TypeError("Dimensions must be iterable of dimensions")
150@dataclasses.dataclass(frozen=True)
151class BaseInput(DimensionedConnection):
152 """Class used for declaring PipelineTask input connections.
154 Attributes
155 ----------
156 name : `str`
157 The default name used to identify the dataset type.
158 storageClass : `str`
159 The storage class used when (un)/persisting the dataset type.
160 multiple : `bool`
161 Indicates if this connection should expect to contain multiple objects
162 of the given dataset type. Tasks with more than one connection with
163 ``multiple=True`` with the same dimensions may want to implement
164 `.PipelineTaskConnections.adjustQuantum` to ensure those datasets are
165 consistent (i.e. zip-iterable) in `PipelineTask.runQuantum` and notify
166 the execution system as early as possible of outputs that will not be
167 produced because the corresponding input is missing.
168 dimensions : iterable of `str`
169 The `lsst.daf.butler.Butler` `lsst.daf.butler.Registry` dimensions used
170 to identify the dataset type identified by the specified name.
171 deferLoad : `bool`
172 Indicates that this dataset type will be loaded as a
173 `lsst.daf.butler.DeferredDatasetHandle`. PipelineTasks can use this
174 object to load the object at a later time.
175 minimum : `bool`
176 Minimum number of datasets required for this connection, per quantum.
177 This is checked in the base implementation of
178 `.PipelineTaskConnections.adjustQuantum`, which raises `NoWorkFound` if
179 the minimum is not met for `Input` connections (causing the quantum to
180 be pruned, skipped, or never created, depending on the context), and
181 `FileNotFoundError` for `PrerequisiteInput` connections (causing
182 QuantumGraph generation to fail). `PipelineTask` implementations may
183 provide custom `~.PipelineTaskConnections.adjustQuantum`
184 implementations for more fine-grained or configuration-driven
185 constraints, as long as they are compatible with this minium.
187 Raises
188 ------
189 TypeError
190 Raised if ``minimum`` is greater than one but ``multiple=False``.
191 NotImplementedError
192 Raised if ``minimum`` is zero for a regular `Input` connection; this
193 is not currently supported by our QuantumGraph generation algorithm.
194 """
196 deferLoad: bool = False
197 minimum: int = 1
199 def __post_init__(self) -> None:
200 super().__post_init__()
201 if self.minimum > 1 and not self.multiple: 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 raise TypeError(f"Cannot set minimum={self.minimum} if multiple=False.")
205@dataclasses.dataclass(frozen=True)
206class Input(BaseInput):
207 """Class used for declaring PipelineTask input connections.
209 Attributes
210 ----------
211 name : `str`
212 The default name used to identify the dataset type.
213 storageClass : `str`
214 The storage class used when (un)/persisting the dataset type.
215 multiple : `bool`
216 Indicates if this connection should expect to contain multiple objects
217 of the given dataset type. Tasks with more than one connection with
218 ``multiple=True`` with the same dimensions may want to implement
219 `.PipelineTaskConnections.adjustQuantum` to ensure those datasets are
220 consistent (i.e. zip-iterable) in `PipelineTask.runQuantum` and notify
221 the execution system as early as possible of outputs that will not be
222 produced because the corresponding input is missing.
223 dimensions : iterable of `str`
224 The `lsst.daf.butler.Butler` `lsst.daf.butler.Registry` dimensions used
225 to identify the dataset type identified by the specified name.
226 deferLoad : `bool`
227 Indicates that this dataset type will be loaded as a
228 `lsst.daf.butler.DeferredDatasetHandle`. PipelineTasks can use this
229 object to load the object at a later time.
230 minimum : `bool`
231 Minimum number of datasets required for this connection, per quantum.
232 This is checked in the base implementation of
233 `.PipelineTaskConnections.adjustQuantum`, which raises `NoWorkFound` if
234 the minimum is not met for `Input` connections (causing the quantum to
235 be pruned, skipped, or never created, depending on the context), and
236 `FileNotFoundError` for `PrerequisiteInput` connections (causing
237 QuantumGraph generation to fail). `PipelineTask` implementations may
238 provide custom `~.PipelineTaskConnections.adjustQuantum`
239 implementations for more fine-grained or configuration-driven
240 constraints, as long as they are compatible with this minium.
241 deferGraphConstraint : `bool`, optional
242 If `True`, do not include this dataset type's existence in the initial
243 query that starts the QuantumGraph generation process. This can be
244 used to make QuantumGraph generation faster by avoiding redundant
245 datasets, and in certain cases it can (along with careful attention to
246 which tasks are included in the same QuantumGraph) be used to work
247 around the QuantumGraph generation algorithm's inflexible handling of
248 spatial overlaps. This option has no effect when the connection is not
249 an overall input of the pipeline (or subset thereof) for which a graph
250 is being created, and it never affects the ordering of quanta.
251 deferBinding : `bool`, optional
252 If `True`, the dataset will not be automatically included in
253 the pipeline graph, ``deferGraphConstraint`` is implied.
254 The custom QuantumGraphBuilder is required to bind it and add a
255 corresponding edge to the pipeline graph.
256 This option allows to have the same dataset type as both
257 input and output of a quantum.
259 Raises
260 ------
261 TypeError
262 Raised if ``minimum`` is greater than one but ``multiple=False``.
263 NotImplementedError
264 Raised if ``minimum`` is zero for a regular `Input` connection; this
265 is not currently supported by our QuantumGraph generation algorithm.
266 """
268 deferGraphConstraint: bool = False
270 deferBinding: bool = False
272 _connection_type_set: ClassVar[str] = "inputs"
275@dataclasses.dataclass(frozen=True)
276class PrerequisiteInput(BaseInput):
277 """Class used for declaring PipelineTask prerequisite connections.
279 Attributes
280 ----------
281 name : `str`
282 The default name used to identify the dataset type.
283 storageClass : `str`
284 The storage class used when (un)/persisting the dataset type.
285 multiple : `bool`
286 Indicates if this connection should expect to contain multiple objects
287 of the given dataset type. Tasks with more than one connection with
288 ``multiple=True`` with the same dimensions may want to implement
289 `.PipelineTaskConnections.adjustQuantum` to ensure those datasets are
290 consistent (i.e. zip-iterable) in `PipelineTask.runQuantum` and notify
291 the execution system as early as possible of outputs that will not be
292 produced because the corresponding input is missing.
293 dimensions : iterable of `str`
294 The `lsst.daf.butler.Butler` `lsst.daf.butler.Registry` dimensions used
295 to identify the dataset type identified by the specified name.
296 minimum : `bool`
297 Minimum number of datasets required for this connection, per quantum.
298 This is checked in the base implementation of
299 `.PipelineTaskConnections.adjustQuantum`, which raises
300 `FileNotFoundError` (causing QuantumGraph generation to fail).
301 `PipelineTask` implementations may provide custom
302 `~.PipelineTaskConnections.adjustQuantum` implementations for more
303 fine-grained or configuration-driven constraints, as long as they are
304 compatible with this minium.
305 lookupFunction : `typing.Callable`, optional
306 An optional callable function that will look up PrerequisiteInputs
307 using the DatasetType, registry, quantum dataId, and input collections
308 passed to it. If no function is specified, the default temporal spatial
309 lookup will be used.
311 Raises
312 ------
313 TypeError
314 Raised if ``minimum`` is greater than one but ``multiple=False``.
316 Notes
317 -----
318 Prerequisite inputs are used for datasets that must exist in the data
319 repository before a pipeline including this is run; they cannot be produced
320 by another task in the same pipeline.
322 In exchange for this limitation, they have a number of advantages relative
323 to regular `Input` connections:
325 - The query used to find them then during `QuantumGraph` generation can be
326 fully customized by providing a ``lookupFunction``.
327 - Failed searches for prerequisites during `QuantumGraph` generation will
328 usually generate more helpful diagnostics than those for regular `Input`
329 connections.
330 - The default query for prerequisite inputs relates the quantum dimensions
331 directly to the dimensions of its dataset type, without being constrained
332 by any of the other dimensions in the pipeline. This allows them to be
333 used for temporal calibration lookups (which regular `Input` connections
334 cannot do at present) and to work around `QuantumGraph` generation
335 limitations involving cases where naive spatial overlap relationships
336 between dimensions are not desired (e.g. a task that wants all detectors
337 in each visit for which the visit overlaps a tract, not just those where
338 that detector+visit combination overlaps the tract).
339 - Prerequisite inputs may be optional (regular inputs are never optional).
340 """
342 lookupFunction: (
343 Callable[[DatasetType, Registry, DataCoordinate, Sequence[str]], Iterable[DatasetRef]] | None
344 ) = None
346 _connection_type_set: ClassVar[str] = "prerequisiteInputs"
349@dataclasses.dataclass(frozen=True)
350class Output(DimensionedConnection):
351 """Connection for output dataset."""
353 _connection_type_set: ClassVar[str] = "outputs"
356@dataclasses.dataclass(frozen=True)
357class InitInput(BaseConnection):
358 """Connection for initInput dataset."""
360 _connection_type_set: ClassVar[str] = "initInputs"
363@dataclasses.dataclass(frozen=True)
364class InitOutput(BaseConnection):
365 """Connection for initOutput dataset."""
367 _connection_type_set: ClassVar[str] = "initOutputs"