Coverage for python/lsst/analysis/tools/interfaces/_analysisTools.py: 76%
181 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-11 09:02 +0000
1# This file is part of analysis_tools.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://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 <https://www.gnu.org/licenses/>.
22from __future__ import annotations
24__all__ = ("AnalysisTool",)
26from collections import ChainMap
27from collections.abc import Callable, Iterable, Mapping
28from functools import wraps
29from operator import attrgetter
30from typing import Protocol, runtime_checkable
32import lsst.pex.config as pexConfig
33from lsst.obs.base import Instrument
34from lsst.pex.config import Field, FieldValidationError, ListField
35from lsst.pex.config.configurableActions import ConfigurableActionField
36from lsst.pipe.base import Pipeline
37from lsst.verify import Measurement
39from ._actions import AnalysisAction, JointAction, JointResults, NoPlot, PlotAction
40from ._interfaces import KeyedData, KeyedDataSchema, KeyedResults, PlotTypes
41from ._stages import BasePrep, BaseProcess, BaseProduce
44@runtime_checkable
45class _HasOutputNames(Protocol):
46 def getOutputNames(self, config: pexConfig.Config | None = None) -> Iterable[str]: ...
49def _finalizeWrapper(
50 f: Callable[[AnalysisTool], None], cls: type[AnalysisTool]
51) -> Callable[[AnalysisTool], None]:
52 """Wrap a classes finalize function to ensure the base classes special
53 finalize method only fires after the most derived finalize method.
55 Parameters
56 ----------
57 f : `Callable`
58 Function that is being wrapped
59 cls : `type` of `AnalysisTool`
60 The class which is having its function wrapped
62 Returns
63 -------
64 function : `Callable`
65 The new function which wraps the old
66 """
68 @wraps(f)
69 def wrapper(self: AnalysisTool) -> None:
70 # call the wrapped finalize function
71 f(self)
72 # get the method resolution order for the self variable
73 mro = self.__class__.mro()
75 # Find which class in the mro that last defines a finalize method
76 # note that this is in the reverse order from the mro, because the
77 # last class in an inheritance stack is the first in the mro (aka you
78 # walk from the furthest child first.
79 #
80 # Also note that the most derived finalize method need not be the same
81 # as the type of self, as that might inherit from a parent somewhere
82 # between it and the furthest parent.
83 mostDerived: type | None = None
84 for klass in mro: 84 ↛ 94line 84 didn't jump to line 94 because the loop on line 84 didn't complete
85 # inspect the classes dictionary to see if it specifically defines
86 # finalize. This is needed because normal lookup will go through
87 # the mro, but this needs to be restricted to each class.
88 if "finalize" in vars(klass):
89 mostDerived = klass
90 break
92 # Find what stage in the MRO walking process the recursive function
93 # call is in.
94 this = super(cls, self).__thisclass__
96 # If the current place in the MRO walking is also the class that
97 # defines the most derived instance of finalize, then call the base
98 # classes private finalize that must be called after everything else.
99 if mostDerived is not None and this == mostDerived:
100 self._baseFinalize()
102 return wrapper
105class AnalysisTool(AnalysisAction):
106 r"""A tool which which calculates a single type of analysis on input data,
107 though it may return more than one result.
109 Although `AnalysisTool`\ s are considered a single type of analysis, the
110 classes themselves can be thought of as a container. `AnalysisTool`\ s
111 are aggregations of `AnalysisAction`\ s to form prep, process, and
112 produce stages. These stages allow better reuse of individual
113 `AnalysisActions` and easier introspection in contexts such as a notebook
114 or interpreter.
116 An `AnalysisTool` can be thought of an an individual configuration that
117 specifies which `AnalysisAction` should run for each stage.
119 The stages themselves are also configurable, allowing control over various
120 aspects of the individual `AnalysisAction`\ s.
121 """
123 prep = ConfigurableActionField[AnalysisAction](doc="Action to run to prepare inputs", default=BasePrep)
124 process = ConfigurableActionField[AnalysisAction](
125 doc="Action to process data into intended form", default=BaseProcess
126 )
127 produce = ConfigurableActionField[AnalysisAction](
128 doc="Action to perform any finalization steps", default=BaseProduce
129 )
130 metric_tags = ListField[str](
131 doc="List of tags which will be associated with metric measurement(s)", default=[]
132 )
134 def __init_subclass__(cls: type[AnalysisTool], **kwargs):
135 super().__init_subclass__(**kwargs)
136 # Wrap all definitions of the finalize method in a special wrapper that
137 # ensures that the bases classes private finalize is called last.
138 if "finalize" in vars(cls):
139 cls.finalize = _finalizeWrapper(cls.finalize, cls)
141 dynamicOutputNames: bool | Field[bool] = False
142 """Determines whether to grant the ``getOutputNames`` method access to
143 config parameters.
144 """
146 parameterizedBand: bool | Field[bool] = True
147 """Specifies if an `AnalysisTool` may parameterize a band within any field
148 in any stage, or if the set of bands is already uniquely determined though
149 configuration. I.e. can this `AnalysisTool` be automatically looped over to
150 produce a result for multiple bands.
151 """
153 propagateData: bool | Field[bool] = False
154 """If this value is set to True, the input data `KeyedData` will be passed
155 to each stage in addition to the ``prep`` stage. Any keys created in a
156 stage with the same key that exists in the input ``data`` will shadow that
157 key/value.
158 """
160 doProducePlots = Field[bool](
161 doc="""If this value is set to False, then no plots are generated.
162 Metric generation is unaffected. If the `AnalysisTool` produces no
163 metrics, then no outputs are generated""",
164 default=True,
165 )
167 def __call__(self, data: KeyedData, **kwargs) -> KeyedResults:
168 bands = kwargs.pop("bands", None)
169 if "plotInfo" in kwargs and kwargs.get("plotInfo") is not None:
170 if "plotName" not in kwargs["plotInfo"] or kwargs["plotInfo"]["plotName"] is None:
171 kwargs["plotInfo"]["plotName"] = self.identity
172 if not self.parameterizedBand or bands is None: 172 ↛ 178line 172 didn't jump to line 178 because the condition on line 172 was always true
173 if "band" not in kwargs:
174 # Some tasks require a "band" key for naming. This shouldn't
175 # affect the results. DM-35813 should make this unnecessary.
176 kwargs["band"] = "analysisTools"
177 return self._call_single(data, **kwargs)
178 results: KeyedResults = {}
179 for band in bands:
180 kwargs["band"] = band
181 if "plotInfo" in kwargs:
182 kwargs["plotInfo"]["bands"] = band
183 subResult = self._call_single(data, **kwargs)
184 for key, value in subResult.items():
185 match value:
186 case PlotTypes():
187 results[f"{band}_{key}"] = value
188 case Measurement():
189 results[key] = value
190 return results
192 def _call_single(self, data: KeyedData, **kwargs) -> KeyedResults:
193 # create a shallow copy of kwargs
194 kwargs = dict(**kwargs)
195 kwargs["metric_tags"] = list(self.metric_tags or ())
196 kwargs["produce_plots"] = self.doProducePlots
197 prepped: KeyedData = self.prep(data, **kwargs) # type: ignore
198 if self.propagateData: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true
199 prepped = ChainMap(data, prepped)
200 processed: KeyedData = self.process(prepped, **kwargs) # type: ignore
201 if self.propagateData: 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 processed = ChainMap(data, processed)
203 # If the analysis tool only produces plots, and doProducePlots is False
204 # then don't run produce; return nothing.
205 if not self.doProducePlots and isinstance(self.produce, PlotAction):
206 return {}
207 finalized: (
208 Mapping[str, PlotTypes] | PlotTypes | Mapping[str, Measurement] | Measurement | JointResults
209 ) = self.produce(
210 processed, **kwargs
211 ) # type: ignore
212 return self._process_single_results(finalized)
214 def _getPlotType(self) -> str:
215 match self.produce:
216 case PlotAction():
217 return self.produce.getPlotType()
218 case JointAction(plot=NoPlot()):
219 pass
220 case JointAction(plot=plotter):
221 return plotter.getPlotType()
223 return ""
225 def _process_single_results(
226 self,
227 results: Mapping[str, PlotTypes] | PlotTypes | Mapping[str, Measurement] | Measurement | JointResults,
228 ) -> KeyedResults:
229 accumulation = {}
230 predicate = f"{self.identity}" if self.identity else ""
231 match results:
232 case Mapping():
233 suffix = self._getPlotType()
234 for key, value in results.items():
235 match value:
236 case Measurement():
237 iterable = (predicate, key)
238 case PlotTypes(): 238 ↛ 240line 238 didn't jump to line 240 because the pattern on line 238 always matched
239 iterable = (predicate, key, suffix)
240 case _:
241 raise RuntimeError(f"Unexpected {key=}, {value=} from:\n{self=}")
242 refKey = "_".join(x for x in iterable if x)
243 accumulation[refKey] = value
244 case PlotTypes():
245 suffix = self._getPlotType()
246 refKey = "_".join(x for x in (predicate, suffix) if x)
247 accumulation[refKey] = results
248 case Measurement(): 248 ↛ 249line 248 didn't jump to line 249 because the pattern on line 248 never matched
249 accumulation[f"{predicate}"] = results
250 case JointResults(plot=plotResults, metric=metricResults): 250 ↛ 257line 250 didn't jump to line 257 because the pattern on line 250 always matched
251 if plotResults is not None:
252 subResult = self._process_single_results(plotResults)
253 accumulation.update(subResult)
254 if metricResults is not None:
255 subResult = self._process_single_results(metricResults)
256 accumulation.update(subResult)
257 return accumulation
259 def getInputSchema(self) -> KeyedDataSchema:
260 return self.prep.getInputSchema()
262 def populatePrepFromProcess(self):
263 """Add additional inputs to the prep stage if supported.
265 If the configured prep action supports adding to its input schema,
266 attempt to add the required inputs schema from the process stage to the
267 prep stage.
269 This method will be a no-op if the prep action does not support this
270 feature.
271 """
272 self.prep.addInputSchema(self.process.getInputSchema())
274 def getOutputNames(self, config: pexConfig.Config | None = None) -> Iterable[str]:
275 """Return the names of the plots produced by this analysis tool.
277 If there is a `PlotAction` defined in the produce action, these names
278 will either come from the `PlotAction` if it defines a
279 ``getOutputNames`` method (likely if it returns a mapping of figures),
280 or a default value is used and a single figure is assumed.
282 Parameters
283 ----------
284 config : `lsst.pex.config.Config`, optional
285 Configuration of the task. This is only used if the output naming
286 needs to be config-aware.
288 Returns
289 -------
290 result : `tuple` of `str`
291 Names for each plot produced by this action.
292 """
293 if not self.doProducePlots:
294 return tuple()
295 match self.produce:
296 case JointAction(plot=NoPlot()): 296 ↛ 297line 296 didn't jump to line 297 because the pattern on line 296 never matched
297 return tuple()
298 case _HasOutputNames(): 298 ↛ 300line 298 didn't jump to line 300 because the pattern on line 298 always matched
299 outNames = tuple(self.produce.getOutputNames(config=config))
300 case _:
301 raise ValueError(f"Unsupported Action type {type(self.produce)} for getting output names")
303 results = []
304 suffix = self._getPlotType()
305 if self.parameterizedBand: 305 ↛ 308line 305 didn't jump to line 308 because the condition on line 305 was always true
306 prefix = "_".join(x for x in ("{band}", self.identity) if x)
307 else:
308 prefix = f"{self.identity}" if self.identity else ""
310 if outNames: 310 ↛ 311line 310 didn't jump to line 311 because the condition on line 310 was never true
311 for name in outNames:
312 results.append("_".join(x for x in (prefix, name, suffix) if x))
313 else:
314 results.append("_".join(x for x in (prefix, suffix) if x))
315 return results
317 @classmethod
318 def fromPipeline(
319 cls,
320 pipeline: str | Pipeline,
321 name: str,
322 fullpath: bool = False,
323 instrument: Instrument | str | None = None,
324 ) -> AnalysisTool | None:
325 """Construct an `AnalysisTool` from a definition written in a
326 `~lsst.pipe.base.Pipeline`.
328 Parameters
329 ----------
330 pipeline : `str` or `~lsst.pipe.base.Pipeline`
331 The pipeline to load the `AnalysisTool` from.
332 name : `str`
333 The name of the analysis tool to run. This can either be just the
334 name assigned to the tool, or an absolute name in a config
335 hierarchy.
336 fullpath : `bool`
337 Determines if the name is interpreted as an absolute path in a
338 config hierarchy, or is relative to an `AnalysisTool` ``atools``
339 `~lsst.pex.config.configurableActions.ConfigurableActionStructField`
340 .
341 instrument : `~lsst.daf.butler.instrument.Instrument` or `str` or\
342 `None`
343 Either a derived class object of a `lsst.daf.butler.instrument` or
344 a string corresponding to a fully qualified
345 `lsst.daf.butler.instrument` name or None if no instrument needs
346 specified or if the pipeline contains the instrument in it.
347 Defaults to None.
349 Returns
350 -------
351 tool : `AnalysisTool`
352 The loaded `AnalysisTool` as configured in the pipeline.
354 Raises
355 ------
356 ValueError
357 Raised if the config field specified does not point to an
358 `AnalysisTool`.
359 Raised if an instrument is specified and it conflicts with the
360 pipelines instrument.
361 """
362 if not isinstance(pipeline, Pipeline): 362 ↛ 366line 362 didn't jump to line 366 because the condition on line 362 was always true
363 pipeline = Pipeline.fromFile(pipeline)
364 # If the caller specified an instrument, verify it does not conflict
365 # with the pipelines instrument, and add it to the pipeline
366 if instrument is not None: 366 ↛ 367line 366 didn't jump to line 367 because the condition on line 366 was never true
367 if (pipeInstrument := pipeline.getInstrument()) and pipeInstrument != instrument:
368 raise ValueError(
369 f"The supplied instrument {instrument} conflicts with the pipelines instrument "
370 f"{pipeInstrument}."
371 )
372 else:
373 pipeline.addInstrument(instrument)
374 try:
375 pipelineGraph = pipeline.to_graph()
376 except (FieldValidationError, ValueError) as err:
377 raise ValueError(
378 "There was an error instantiating the pipeline, do you need to specify an instrument?"
379 ) from err
380 if not fullpath:
381 name = f"atools.{name}"
382 for task in pipelineGraph.tasks.values(): 382 ↛ 391line 382 didn't jump to line 391 because the loop on line 382 didn't complete
383 config = task.config
384 try:
385 attr = attrgetter(name)(config)
386 except AttributeError:
387 continue
388 if not isinstance(attr, AnalysisTool): 388 ↛ 389line 388 didn't jump to line 389 because the condition on line 388 was never true
389 raise ValueError("The requested name did not refer to an analysisTool")
390 return attr
391 return None
393 def finalize(self) -> None:
394 """Run any finalization code that depends on configuration being
395 complete.
396 """
397 pass
399 def _baseFinalize(self) -> None:
400 self.populatePrepFromProcess()
402 def freeze(self):
403 if not self.__dict__.get("_finalizeRun"): 403 ↛ 406line 403 didn't jump to line 406 because the condition on line 403 was always true
404 self.finalize()
405 self.__dict__["_finalizeRun"] = True
406 super().freeze()
409# explicitly wrap the finalize of the base class
410AnalysisTool.finalize = _finalizeWrapper(AnalysisTool.finalize, AnalysisTool)