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
pipelineTask.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 """This module defines PipelineTask class and related methods.
23 """
24 
25 __all__ = ["PipelineTask"] # Classes in this module
26 
27 from .task import Task
28 from .butlerQuantumContext import ButlerQuantumContext
29 from .connections import InputQuantizedConnection, OutputQuantizedConnection
30 
31 
33  """Base class for all pipeline tasks.
34 
35  This is an abstract base class for PipelineTasks which represents an
36  algorithm executed by framework(s) on data which comes from data butler,
37  resulting data is also stored in a data butler.
38 
39  PipelineTask inherits from a `pipe.base.Task` and uses the same
40  configuration mechanism based on `pex.config`. `PipelineTask` classes also
41  have a `PipelineTaskConnections` class associated with their config which
42  defines all of the IO a `PipelineTask` will need to do. PipelineTask
43  sub-class typically implements `run()` method which receives Python-domain
44  data objects and returns `pipe.base.Struct` object with resulting data.
45  `run()` method is not supposed to perform any I/O, it operates entirely on
46  in-memory objects. `runQuantum()` is the method (can be re-implemented in
47  sub-class) where all necessary I/O is performed, it reads all input data
48  from data butler into memory, calls `run()` method with that data, examines
49  returned `Struct` object and saves some or all of that data back to data
50  butler. `runQuantum()` method receives a `ButlerQuantumContext` instance to
51  facilitate I/O, a `InputQuantizedConnection` instance which defines all
52  input `lsst.daf.butler.DatasetRef`, and a `OutputQuantizedConnection`
53  instance which defines all the output `lsst.daf.butler.DatasetRef` for a
54  single invocation of PipelineTask.
55 
56  Subclasses must be constructable with exactly the arguments taken by the
57  PipelineTask base class constructor, but may support other signatures as
58  well.
59 
60  Attributes
61  ----------
62  canMultiprocess : bool, True by default (class attribute)
63  This class attribute is checked by execution framework, sub-classes
64  can set it to ``False`` in case task does not support multiprocessing.
65 
66  Parameters
67  ----------
68  config : `pex.config.Config`, optional
69  Configuration for this task (an instance of ``self.ConfigClass``,
70  which is a task-specific subclass of `PipelineTaskConfig`).
71  If not specified then it defaults to `self.ConfigClass()`.
72  log : `lsst.log.Log`, optional
73  Logger instance whose name is used as a log name prefix, or ``None``
74  for no prefix.
75  initInputs : `dict`, optional
76  A dictionary of objects needed to construct this PipelineTask, with
77  keys matching the keys of the dictionary returned by
78  `getInitInputDatasetTypes` and values equivalent to what would be
79  obtained by calling `Butler.get` with those DatasetTypes and no data
80  IDs. While it is optional for the base class, subclasses are
81  permitted to require this argument.
82  """
83  canMultiprocess = True
84 
85  def __init__(self, *, config=None, log=None, initInputs=None, **kwargs):
86  super().__init__(config=config, log=log, **kwargs)
87 
88  def run(self, **kwargs):
89  """Run task algorithm on in-memory data.
90 
91  This method should be implemented in a subclass. This method will
92  receive keyword arguments whose names will be the same as names of
93  connection fields describing input dataset types. Argument values will
94  be data objects retrieved from data butler. If a dataset type is
95  configured with ``multiple`` field set to ``True`` then the argument
96  value will be a list of objects, otherwise it will be a single object.
97 
98  If the task needs to know its input or output DataIds then it has to
99  override `runQuantum` method instead.
100 
101  This method should return a `Struct` whose attributes share the same
102  name as the connection fields describing output dataset types.
103 
104  Returns
105  -------
106  struct : `Struct`
107  Struct with attribute names corresponding to output connection
108  fields
109 
110  Examples
111  --------
112  Typical implementation of this method may look like:
113 
114  .. code-block:: python
115 
116  def run(self, input, calib):
117  # "input", "calib", and "output" are the names of the config
118  # fields
119 
120  # Assuming that input/calib datasets are `scalar` they are
121  # simple objects, do something with inputs and calibs, produce
122  # output image.
123  image = self.makeImage(input, calib)
124 
125  # If output dataset is `scalar` then return object, not list
126  return Struct(output=image)
127 
128  """
129  raise NotImplementedError("run() is not implemented")
130 
131  def runQuantum(self, butlerQC: ButlerQuantumContext, inputRefs: InputQuantizedConnection,
132  outputRefs: OutputQuantizedConnection):
133  """Method to do butler IO and or transforms to provide in memory
134  objects for tasks run method
135 
136  Parameters
137  ----------
138  butlerQC : `ButlerQuantumContext`
139  A butler which is specialized to operate in the context of a
140  `lsst.daf.butler.Quantum`.
141  inputRefs : `InputQuantizedConnection`
142  Datastructure whose attribute names are the names that identify
143  connections defined in corresponding `PipelineTaskConnections`
144  class. The values of these attributes are the
145  `lsst.daf.butler.DatasetRef` objects associated with the defined
146  input/prerequisite connections.
147  outputRefs : `OutputQuantizedConnection`
148  Datastructure whose attribute names are the names that identify
149  connections defined in corresponding `PipelineTaskConnections`
150  class. The values of these attributes are the
151  `lsst.daf.butler.DatasetRef` objects associated with the defined
152  output connections.
153  """
154  inputs = butlerQC.get(inputRefs)
155  outputs = self.runrun(**inputs)
156  butlerQC.put(outputs, outputRefs)
157 
158  def getResourceConfig(self):
159  """Return resource configuration for this task.
160 
161  Returns
162  -------
163  Object of type `~config.ResourceConfig` or ``None`` if resource
164  configuration is not defined for this task.
165  """
166  return getattr(self.configconfig, "resources", None)
def __init__(self, *config=None, log=None, initInputs=None, **kwargs)
Definition: pipelineTask.py:85
def runQuantum(self, ButlerQuantumContext butlerQC, InputQuantizedConnection inputRefs, OutputQuantizedConnection outputRefs)