LSST Applications  21.0.0+04719a4bac,21.0.0-1-ga51b5d4+f5e6047307,21.0.0-11-g2b59f77+a9c1acf22d,21.0.0-11-ga42c5b2+86977b0b17,21.0.0-12-gf4ce030+76814010d2,21.0.0-13-g1721dae+760e7a6536,21.0.0-13-g3a573fe+768d78a30a,21.0.0-15-g5a7caf0+f21cbc5713,21.0.0-16-g0fb55c1+b60e2d390c,21.0.0-19-g4cded4ca+71a93a33c0,21.0.0-2-g103fe59+bb20972958,21.0.0-2-g45278ab+04719a4bac,21.0.0-2-g5242d73+3ad5d60fb1,21.0.0-2-g7f82c8f+8babb168e8,21.0.0-2-g8f08a60+06509c8b61,21.0.0-2-g8faa9b5+616205b9df,21.0.0-2-ga326454+8babb168e8,21.0.0-2-gde069b7+5e4aea9c2f,21.0.0-2-gecfae73+1d3a86e577,21.0.0-2-gfc62afb+3ad5d60fb1,21.0.0-25-g1d57be3cd+e73869a214,21.0.0-3-g357aad2+ed88757d29,21.0.0-3-g4a4ce7f+3ad5d60fb1,21.0.0-3-g4be5c26+3ad5d60fb1,21.0.0-3-g65f322c+e0b24896a3,21.0.0-3-g7d9da8d+616205b9df,21.0.0-3-ge02ed75+a9c1acf22d,21.0.0-4-g591bb35+a9c1acf22d,21.0.0-4-g65b4814+b60e2d390c,21.0.0-4-gccdca77+0de219a2bc,21.0.0-4-ge8a399c+6c55c39e83,21.0.0-5-gd00fb1e+05fce91b99,21.0.0-6-gc675373+3ad5d60fb1,21.0.0-64-g1122c245+4fb2b8f86e,21.0.0-7-g04766d7+cd19d05db2,21.0.0-7-gdf92d54+04719a4bac,21.0.0-8-g5674e7b+d1bd76f71f,master-gac4afde19b+a9c1acf22d,w.2021.13
LSST Data Management Base Package
butlerQuantumContext.py
Go to the documentation of this file.
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 program is free software: you can redistribute it and/or modify
10 # it under the terms of the GNU General Public License as published by
11 # the Free Software Foundation, either version 3 of the License, or
12 # (at your option) any later version.
13 #
14 # This program is distributed in the hope that it will be useful,
15 # but WITHOUT ANY WARRANTY; without even the implied warranty of
16 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 # GNU General Public License for more details.
18 #
19 # You should have received a copy of the GNU General Public License
20 # along with this program. If not, see <http://www.gnu.org/licenses/>.
21 
22 """Module defining a butler like object specialized to a specific quantum.
23 """
24 
25 __all__ = ("ButlerQuantumContext",)
26 
27 import types
28 import typing
29 
30 from .connections import InputQuantizedConnection, OutputQuantizedConnection, DeferredDatasetRef
31 from .struct import Struct
32 from lsst.daf.butler import DatasetRef, Butler, Quantum
33 
34 
36  """A Butler-like class specialized for a single quantum
37 
38  A ButlerQuantumContext class wraps a standard butler interface and
39  specializes it to the context of a given quantum. What this means
40  in practice is that the only gets and puts that this class allows
41  are DatasetRefs that are contained in the quantum.
42 
43  In the future this class will also be used to record provenance on
44  what was actually get and put. This is in contrast to what the
45  preflight expects to be get and put by looking at the graph before
46  execution.
47 
48  Parameters
49  ----------
50  butler : `lsst.daf.butler.Butler`
51  Butler object from/to which datasets will be get/put
52  quantum : `lsst.daf.butler.core.Quantum`
53  Quantum object that describes the datasets which will be get/put by a
54  single execution of this node in the pipeline graph. All input
55  dataset references must be resolved (i.e. satisfy
56  ``DatasetRef.id is not None``) prior to constructing the
57  `ButlerQuantumContext`.
58 
59  Notes
60  -----
61  Most quanta in any non-trivial graph will not start with resolved dataset
62  references, because they represent processing steps that can only run
63  after some other quanta have produced their inputs. At present, it is the
64  responsibility of ``lsst.ctrl.mpexec.SingleQuantumExecutor`` to resolve all
65  datasets prior to constructing `ButlerQuantumContext` and calling
66  `runQuantum`, and the fact that this precondition is satisfied by code in
67  a downstream package is considered a problem with the
68  ``pipe_base/ctrl_mpexec`` separation of concerns that will be addressed in
69  the future.
70  """
71  def __init__(self, butler: Butler, quantum: Quantum):
72  self.quantumquantum = quantum
73  self.registryregistry = butler.registry
74  self.allInputsallInputs = set()
75  self.allOutputsallOutputs = set()
76  for refs in quantum.inputs.values():
77  for ref in refs:
78  self.allInputsallInputs.add((ref.datasetType, ref.dataId))
79  for refs in quantum.outputs.values():
80  for ref in refs:
81  self.allOutputsallOutputs.add((ref.datasetType, ref.dataId))
82 
83  # Create closures over butler to discourage anyone from directly
84  # using a butler reference
85  def _get(self, ref):
86  # Butler methods below will check for unresolved DatasetRefs and
87  # raise appropriately, so no need for us to do that here.
88  if isinstance(ref, DeferredDatasetRef):
89  self._checkMembership_checkMembership(ref.datasetRef, self.allInputsallInputs)
90  return butler.getDirectDeferred(ref.datasetRef)
91 
92  else:
93  self._checkMembership_checkMembership(ref, self.allInputsallInputs)
94  return butler.getDirect(ref)
95 
96  def _put(self, value, ref):
97  self._checkMembership_checkMembership(ref, self.allOutputsallOutputs)
98  butler.put(value, ref)
99 
100  self._get_get = types.MethodType(_get, self)
101  self._put_put = types.MethodType(_put, self)
102 
103  def get(self, dataset: typing.Union[InputQuantizedConnection,
104  typing.List[DatasetRef],
105  DatasetRef]) -> object:
106  """Fetches data from the butler
107 
108  Parameters
109  ----------
110  dataset
111  This argument may either be an `InputQuantizedConnection` which
112  describes all the inputs of a quantum, a list of
113  `~lsst.daf.butler.DatasetRef`, or a single
114  `~lsst.daf.butler.DatasetRef`. The function will get and return
115  the corresponding datasets from the butler.
116 
117  Returns
118  -------
119  return : `object`
120  This function returns arbitrary objects fetched from the bulter.
121  The structure these objects are returned in depends on the type of
122  the input argument. If the input dataset argument is a
123  `InputQuantizedConnection`, then the return type will be a
124  dictionary with keys corresponding to the attributes of the
125  `InputQuantizedConnection` (which in turn are the attribute
126  identifiers of the connections). If the input argument is of type
127  `list` of `~lsst.daf.butler.DatasetRef` then the return type will
128  be a list of objects. If the input argument is a single
129  `~lsst.daf.butler.DatasetRef` then a single object will be
130  returned.
131 
132  Raises
133  ------
134  ValueError
135  Raised if a `DatasetRef` is passed to get that is not defined in
136  the quantum object
137  """
138  if isinstance(dataset, InputQuantizedConnection):
139  retVal = {}
140  for name, ref in dataset:
141  if isinstance(ref, list):
142  val = [self._get_get(r) for r in ref]
143  else:
144  val = self._get_get(ref)
145  retVal[name] = val
146  return retVal
147  elif isinstance(dataset, list):
148  return [self._get_get(x) for x in dataset]
149  elif isinstance(dataset, DatasetRef) or isinstance(dataset, DeferredDatasetRef):
150  return self._get_get(dataset)
151  else:
152  raise TypeError("Dataset argument is not a type that can be used to get")
153 
154  def put(self, values: typing.Union[Struct, typing.List[typing.Any], object],
155  dataset: typing.Union[OutputQuantizedConnection, typing.List[DatasetRef], DatasetRef]):
156  """Puts data into the butler
157 
158  Parameters
159  ----------
160  values : `Struct` or `list` of `object` or `object`
161  The data that should be put with the butler. If the type of the
162  dataset is `OutputQuantizedConnection` then this argument should be
163  a `Struct` with corresponding attribute names. Each attribute
164  should then correspond to either a list of object or a single
165  object depending of the type of the corresponding attribute on
166  dataset. I.e. if ``dataset.calexp`` is
167  ``[datasetRef1, datasetRef2]`` then ``values.calexp`` should be
168  ``[calexp1, calexp2]``. Like wise if there is a single ref, then
169  only a single object need be passed. The same restriction applies
170  if dataset is directly a `list` of `DatasetRef` or a single
171  `DatasetRef`.
172  dataset
173  This argument may either be an `InputQuantizedConnection` which
174  describes all the inputs of a quantum, a list of
175  `lsst.daf.butler.DatasetRef`, or a single
176  `lsst.daf.butler.DatasetRef`. The function will get and return
177  the corresponding datasets from the butler.
178 
179  Raises
180  ------
181  ValueError
182  Raised if a `DatasetRef` is passed to put that is not defined in
183  the quantum object, or the type of values does not match what is
184  expected from the type of dataset.
185  """
186  if isinstance(dataset, OutputQuantizedConnection):
187  if not isinstance(values, Struct):
188  raise ValueError("dataset is a OutputQuantizedConnection, a Struct with corresponding"
189  " attributes must be passed as the values to put")
190  for name, refs in dataset:
191  valuesAttribute = getattr(values, name)
192  if isinstance(refs, list):
193  if len(refs) != len(valuesAttribute):
194  raise ValueError(f"There must be a object to put for every Dataset ref in {name}")
195  for i, ref in enumerate(refs):
196  self._put_put(valuesAttribute[i], ref)
197  else:
198  self._put_put(valuesAttribute, refs)
199  elif isinstance(dataset, list):
200  if len(dataset) != len(values):
201  raise ValueError("There must be a common number of references and values to put")
202  for i, ref in enumerate(dataset):
203  self._put_put(values[i], ref)
204  elif isinstance(dataset, DatasetRef):
205  self._put_put(values, dataset)
206  else:
207  raise TypeError("Dataset argument is not a type that can be used to put")
208 
209  def _checkMembership(self, ref: typing.Union[typing.List[DatasetRef], DatasetRef], inout: set):
210  """Internal function used to check if a DatasetRef is part of the input
211  quantum
212 
213  This function will raise an exception if the ButlerQuantumContext is
214  used to get/put a DatasetRef which is not defined in the quantum.
215 
216  Parameters
217  ----------
218  ref : `list` of `DatasetRef` or `DatasetRef`
219  Either a list or a single `DatasetRef` to check
220  inout : `set`
221  The connection type to check, e.g. either an input or an output.
222  This prevents both types needing to be checked for every operation,
223  which may be important for Quanta with lots of `DatasetRef`.
224  """
225  if not isinstance(ref, list):
226  ref = [ref]
227  for r in ref:
228  if (r.datasetType, r.dataId) not in inout:
229  raise ValueError("DatasetRef is not part of the Quantum being processed")
def _checkMembership(self, typing.Union[typing.List[DatasetRef], DatasetRef] ref, set inout)
def put(self, typing.Union[Struct, typing.List[typing.Any], object] values, typing.Union[OutputQuantizedConnection, typing.List[DatasetRef], DatasetRef] dataset)
object get(self, typing.Union[InputQuantizedConnection, typing.List[DatasetRef], DatasetRef] dataset)
def __init__(self, Butler butler, Quantum quantum)
daf::base::PropertySet * set
Definition: fits.cc:912