Coverage for python/lsst/analysis/tools/interfaces/_task.py: 37%
273 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/>.
21from __future__ import annotations
23__all__ = ("AnalysisBaseConnections", "AnalysisBaseConfig", "AnalysisPipelineTask")
25import datetime
26import logging
27import pprint
28import traceback
29import warnings
30import weakref
31from collections.abc import Collection, Iterable, Mapping, MutableMapping
32from copy import deepcopy
33from typing import TYPE_CHECKING, Any, cast
35import matplotlib.pyplot as plt
37from lsst.verify import Measurement
39if TYPE_CHECKING:
40 from lsst.daf.butler import DatasetRef, DeferredDatasetHandle
41 from lsst.pipe.base import QuantumContext
43from lsst.daf.butler import DataCoordinate
44from lsst.pex.config import Field, ListField
45from lsst.pex.config.configurableActions import ConfigurableActionStructField
46from lsst.pipe.base import (
47 AlgorithmError,
48 AnnotatedPartialOutputsError,
49 Instrument,
50 PipelineTask,
51 PipelineTaskConfig,
52 PipelineTaskConnections,
53 Struct,
54)
55from lsst.pipe.base import connectionTypes as ct
56from lsst.pipe.base.connections import InputQuantizedConnection, OutputQuantizedConnection
57from lsst.pipe.base.pipelineIR import ConfigIR, ParametersIR
59from ._actions import JointAction, MetricAction, NoMetric
60from ._analysisTools import AnalysisTool
61from ._interfaces import KeyedData, PlotTypes
62from ._metricMeasurementBundle import MetricMeasurementBundle
64"""Base class implementation for the classes needed in creating `PipelineTasks`
65which execute `AnalysisTools`.
67The classes defined in this module have all the required behaviors for
68defining, introspecting, and executing `AnalysisTools` against an input dataset
69type.
71Subclasses of these tasks should specify specific datasets to consume in their
72connection classes and should specify a unique name
73"""
75# TODO: This rcParams modification is a temporary solution, hiding
76# a matplotlib warning indicating too many figures have been opened.
77# When DM-39114 is implemented, this should be removed.
78plt.rcParams.update({"figure.max_open_warning": 0})
81# TODO: This _plotCloser function assists in closing all open plots at the
82# conclusion of a PipelineTask. When DM-39114 is implemented, this function and
83# all associated usage thereof should be removed.
84def _plotCloser(*args):
85 """Close all the plots in the given list."""
86 for plot in args:
87 plt.close(plot)
90class PartialAtoolsFailureError(AlgorithmError):
91 """Raised if an analysis tools action has partially failed.
93 Parameters
94 ----------
95 caughtErrors : `dict` of (`str`, `str`)
96 A dictionary mapping the action name to the string representation of
97 the exception raised by that action.
98 """
100 def __init__(self, caughtErrors) -> None:
101 self._caughtErrors = caughtErrors
102 errString = "These tasks have failed for the following reasons: "
103 for errName in caughtErrors.keys():
104 errString += errName + ": "
105 errString += pprint.pformat(caughtErrors[errName])
106 errString += ", "
107 errString = errString[:-2]
108 super().__init__(errString)
110 @property
111 def metadata(self) -> dict:
112 return self._caughtErrors
115class AnalysisBaseConnections(
116 PipelineTaskConnections, dimensions={}, defaultTemplates={"outputName": "Placeholder"}
117):
118 r"""Base class for Connections used for AnalysisTools PipelineTasks.
120 This class has a pre-defined output connection for the
121 MetricMeasurementMapping. The dataset type name for this connection is
122 determined by the template ``outputName``.
124 Output connections for plots created by `AnalysisPlot`\ s are created
125 dynamically when an instance of the class is created. The init method
126 examines all the `AnalysisPlot` actions specified in the associated
127 `AnalysisBaseConfig` subclass accumulating all the info needed to
128 create the output connections.
130 The dimensions for all of the output connections (metric and plot) will
131 be the same as the dimensions specified for the AnalysisBaseConnections
132 subclass (i.e. quantum dimensions).
133 """
135 metrics = ct.Output(
136 doc="Metrics calculated on input dataset type",
137 name="{outputName}_metrics",
138 storageClass="MetricMeasurementBundle",
139 )
141 def __init__(self, *, config: AnalysisBaseConfig = None): # type: ignore
142 # Validate that the outputName template has been set in config. This
143 # should have been checked early with the configs validate method, but
144 # it is possible for someone to manually create everything in a script
145 # without running validate, so also check it late here.
146 if (outputName := config.connections.outputName) == "Placeholder": # type: ignore 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true
147 raise RuntimeError(
148 "Subclasses must specify an alternative value for the defaultTemplate `outputName`"
149 )
150 super().__init__(config=config)
152 # All arguments must be passed by kw, but python has not method to do
153 # that without specifying a default, so None is used. Validate that
154 # it is not None. This is largely for typing reasons, as in the normal
155 # course of operation code execution paths ensure this will not be None
156 assert config is not None
158 for tool in config.atools:
159 match tool.produce:
160 case JointAction(): 160 ↛ 166line 160 didn't jump to line 166 because the pattern on line 160 always matched
161 if isinstance(tool.produce.metric, NoMetric): 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true
162 continue
163 if len(tool.produce.metric.units) != 0: 163 ↛ 164line 163 didn't jump to line 164 because the condition on line 163 was never true
164 hasMetrics = True
165 break
166 case MetricAction():
167 hasMetrics = True
168 break
169 else:
170 hasMetrics = False
172 # Set the dimensions for the metric
173 if hasMetrics: 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true
174 self.metrics = ct.Output(
175 name=self.metrics.name,
176 doc=self.metrics.doc,
177 storageClass=self.metrics.storageClass,
178 dimensions=self.dimensions,
179 multiple=False,
180 isCalibration=False,
181 )
182 else:
183 # There are no metrics to produce, remove the output connection
184 self.outputs.remove("metrics")
186 # Look for any conflicting names, creating a set of them, as these
187 # will be added to the instance as well as recorded in the outputs
188 # set.
189 existingNames = set(dir(self))
191 # Accumulate all the names to be used from all of the defined
192 # AnalysisPlots.
193 names: Mapping[str, AnalysisTool] = {}
194 for action in config.atools:
195 if not action.doProducePlots:
196 continue
197 if action.dynamicOutputNames: 197 ↛ 198line 197 didn't jump to line 198 because the condition on line 197 was never true
198 outNames = action.getOutputNames(config=config)
199 else:
200 outNames = action.getOutputNames()
201 if action.parameterizedBand: 201 ↛ 205line 201 didn't jump to line 205 because the condition on line 201 was always true
202 for band in config.bands:
203 names.update({name.format(band=band): action for name in outNames})
204 else:
205 names.update({name: action for name in outNames})
207 # For each of the names found, create output connections.
208 for name, action in names.items():
209 name = f"{outputName}_{name}"
210 if name in self.outputs or name in existingNames: 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true
211 raise NameError(
212 f"Plot with name {name} conflicts with existing connection are two plots named the same?"
213 )
215 if action.parameterizedBand and "band" not in self.dimensions: 215 ↛ 231line 215 didn't jump to line 231 because the condition on line 215 was always true
216 # If band is in self.dimensions, but the bands still appear
217 # in the output names, then it is likely a user error or the
218 # user is intentional in having a pair-wise plot. So we will
219 # let the band names pass through in the else block. To avoid
220 # the band from appearing, parametrizedBand needs to be False.
221 multiple = True
222 dimensions = self.dimensions.union({"band"})
223 # Struct attributes have the structure
224 # {outputName}_{band}_{name}.
225 band_name = name.split(outputName, maxsplit=1)[1] # _{band}_{name}.
226 # Limit the maximum split to 2 to avoid splitting the name
227 # that may contain underscores.
228 band, name = band_name.split("_", maxsplit=2)[1:] # ["", band, name].
229 name = f"{outputName}_{name}"
230 else:
231 multiple = False
232 dimensions = self.dimensions
234 outConnection = ct.Output(
235 name=name,
236 storageClass="Plot",
237 doc="Dynamic connection for plotting",
238 dimensions=dimensions,
239 multiple=multiple,
240 )
241 setattr(self, name, outConnection)
243 def adjustQuantum(
244 self,
245 inputs: dict[str, tuple[ct.BaseInput, Collection[DatasetRef]]],
246 outputs: dict[str, tuple[ct.Output, Collection[DatasetRef]]],
247 label: str,
248 data_id: DataCoordinate,
249 ) -> tuple[
250 Mapping[str, tuple[ct.BaseInput, Collection[DatasetRef]]],
251 Mapping[str, tuple[ct.Output, Collection[DatasetRef]]],
252 ]:
253 # If the task does not have band in its dimensions but an output
254 # connection does, only keep the output refs that are consistent with
255 # the producing AnalysisTool's 'bands' attribute.
256 if "band" in data_id.dimensions.names:
257 # Nothing to do, since the task dimensions already have band; just
258 # delegate to super. Note that we use the data ID dimensions since
259 # self.dimensions is not an (expanded) DimensionGroup.
260 return super().adjustQuantum(inputs, outputs, label, data_id)
261 adjusted_outputs: dict[str, tuple[ct.Output, list[DatasetRef]]] = {}
262 for name, (connection, old_refs) in outputs.items():
263 # We also don't want to rely on connection.dimensions; the
264 # dimensions in the refs are expanded and hence safer.
265 new_refs: list[DatasetRef] | None = None
266 for old_ref in old_refs:
267 if new_refs is None:
268 if "band" in old_ref.datasetType.dimensions:
269 new_refs = []
270 else:
271 # If this ref doesn't have 'band' none of the others
272 # for this connection will either; move on to the next.
273 break
274 if old_ref.dataId["band"] in self.config.bands:
275 new_refs.append(old_ref)
276 if new_refs is not None:
277 adjusted_outputs[name] = (connection, new_refs)
278 # Update the original outputs so we can pass them to super.
279 outputs.update(adjusted_outputs)
280 super().adjustQuantum(inputs, outputs, label, data_id)
281 # Return only the connections we modified.
282 return {}, adjusted_outputs
285def _timestampValidator(value: str) -> bool:
286 if value in ("reference_package_timestamp", "run_timestamp", "current_timestamp", "dataset_timestamp"): 286 ↛ 288line 286 didn't jump to line 288 because the condition on line 286 was always true
287 return True
288 elif "explicit_timestamp" in value:
289 try:
290 _, splitTime = value.split(":")
291 except ValueError:
292 logging.error(
293 "Explicit timestamp must be given in the format 'explicit_timestamp:datetime', "
294 r"where datetime is given in the form '%Y%m%dT%H%M%S%z"
295 )
296 return False
297 try:
298 datetime.datetime.strptime(splitTime, r"%Y%m%dT%H%M%S%z")
299 except ValueError:
300 # This is explicitly chosen to be an f string as the string
301 # contains control characters.
302 logging.error(
303 f"The supplied datetime {splitTime} could not be parsed correctly into "
304 r"%Y%m%dT%H%M%S%z format"
305 )
306 return False
307 return True
308 else:
309 return False
312class AnalysisBaseConfig(PipelineTaskConfig, pipelineConnections=AnalysisBaseConnections):
313 """Base class for all configs used to define an `AnalysisPipelineTask`.
315 This base class defines two fields that should be used in all subclasses,
316 atools, and bands.
318 The ``atools`` field is where the user configures which analysis tools will
319 be run as part of this `PipelineTask`.
321 The bands field specifies which bands will be looped over for
322 `AnalysisTools` which support parameterized bands. I.e. called once for
323 each band in the list.
324 """
326 atools = ConfigurableActionStructField[AnalysisTool](
327 doc="The analysis tools that are to be run by this task at execution"
328 )
329 # Temporarally alias these for backwards compatibility
330 plots = atools
331 metrics = atools
332 bands = ListField[str](
333 doc="Filter bands on which to run all of the actions", default=["u", "g", "r", "i", "z", "y"]
334 )
335 metric_tags = ListField[str](
336 doc="List of tags which will be added to all configurable actions", default=[]
337 )
338 doProducePlots = Field[bool](
339 doc="Produce plot outputs for the atools configured on this "
340 "task. Set to False to disable plot generation for all atools "
341 "run by this task while still producing their metrics.",
342 default=True,
343 )
344 dataset_identifier = Field[str](doc="An identifier to be associated with output Metrics", optional=True)
345 reference_package = Field[str](
346 doc="A package who's version, at the time of metric upload to a "
347 "time series database, will be converted to a timestamp of when "
348 "that version was produced",
349 default="lsst_distrib",
350 )
351 timestamp_version = Field[str](
352 doc="Which time stamp should be used as the reference timestamp for a "
353 "metric in a time series database, valid values are; "
354 "reference_package_timestamp, run_timestamp, current_timestamp, "
355 "dataset_timestamp and explicit_timestamp:datetime where datetime is "
356 "given in the form %Y%m%dT%H%M%S%z",
357 default="run_timestamp",
358 check=_timestampValidator,
359 )
360 addOutputNamePrefix = Field[bool](
361 doc="If True, the connections class output name will be prefixed "
362 "to the analysis 'atools' name when the value is dispatched to sasquatch.",
363 default=False,
364 )
366 def applyConfigOverrides(
367 self,
368 instrument: Instrument | None,
369 taskDefaultName: str,
370 pipelineConfigs: Iterable[ConfigIR] | None,
371 parameters: ParametersIR,
372 label: str,
373 ) -> None:
374 extraConfig = {}
375 if (value := parameters.mapping.get("sasquatch_dataset_identifier", None)) is not None: 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true
376 extraConfig["dataset_identifier"] = value
377 if (value := parameters.mapping.get("sasquatch_reference_package", None)) is not None: 377 ↛ 378line 377 didn't jump to line 378 because the condition on line 377 was never true
378 extraConfig["reference_package"] = value
379 if (value := parameters.mapping.get("sasquatch_timestamp_version", None)) is not None: 379 ↛ 380line 379 didn't jump to line 380 because the condition on line 379 was never true
380 if "explicit_timestamp" in value:
381 try:
382 _, splitTime = value.split(":")
383 except ValueError as excpt:
384 raise ValueError(
385 "Explicit timestamp must be given in the format 'explicit_timestamp:datetime', "
386 "where datetime is given in the form '%Y%m%dT%H%M%S%z"
387 ) from excpt
388 try:
389 datetime.datetime.strptime(splitTime, r"%Y%m%dT%H%M%S%z")
390 except ValueError as excpt:
391 raise ValueError(
392 f"The supplied datetime {splitTime} could not be parsed correctly into "
393 "%Y%m%dT%H%M%S%z format"
394 ) from excpt
395 extraConfig["timestamp_version"] = value
396 if extraConfig: 396 ↛ 397line 396 didn't jump to line 397 because the condition on line 396 was never true
397 newPipelineConfigs = [ConfigIR(rest=extraConfig)]
398 if pipelineConfigs is not None:
399 newPipelineConfigs.extend(pipelineConfigs)
400 pipelineConfigs = newPipelineConfigs
401 return super().applyConfigOverrides(instrument, taskDefaultName, pipelineConfigs, parameters, label)
403 def freeze(self):
404 # Copy the meta configuration values to each of the configured tools
405 # only do this if the tool has not been further specialized
406 if not self._frozen: 406 ↛ 412line 406 didn't jump to line 412 because the condition on line 406 was always true
407 for tool in self.atools:
408 for tag in self.metric_tags: 408 ↛ 409line 408 didn't jump to line 409 because the loop on line 408 never started
409 tool.metric_tags.insert(-1, tag)
410 if not self.doProducePlots:
411 tool.doProducePlots = False
412 super().freeze()
414 def validate(self):
415 super().validate()
416 # Validate that the required connections template is set.
417 if self.connections.outputName == "Placeholder": # type: ignore 417 ↛ 418line 417 didn't jump to line 418 because the condition on line 417 was never true
418 raise RuntimeError("Connections class 'outputName' must have a config explicitly set")
421class _StandinPlotInfo(dict):
422 """This class is an implementation detail to support plots in the instance
423 no PlotInfo object is present in the call to run.
424 """
426 def __missing__(self, key):
427 return ""
430class AnalysisPipelineTask(PipelineTask):
431 """Base class for `PipelineTasks` intended to run `AnalysisTools`.
433 The run method will run all of the `AnalysisTools` defined in the config
434 class.
435 """
437 # Typing config because type checkers dont know about our Task magic
438 config: AnalysisBaseConfig
439 ConfigClass = AnalysisBaseConfig
441 warnings_all = (
442 "divide by zero encountered in divide",
443 "invalid value encountered in arcsin",
444 "invalid value encountered in cos",
445 "invalid value encountered in divide",
446 "invalid value encountered in log10",
447 "invalid value encountered in scalar divide",
448 "invalid value encountered in sin",
449 "invalid value encountered in sqrt",
450 "invalid value encountered in true_divide",
451 "Mean of empty slice",
452 )
454 def _runTools(self, data: KeyedData, **kwargs) -> Struct:
455 caughtErrors = {}
456 with warnings.catch_warnings():
457 # Change below to "in self.warnings_all" to find otherwise
458 # unfiltered numpy warnings.
459 for warning in ():
460 warnings.filterwarnings("error", warning, RuntimeWarning)
461 results = Struct()
462 prefixName = f"{self.config.connections.outputName}_" if self.config.addOutputNamePrefix else ""
463 results.metrics = MetricMeasurementBundle(
464 dataset_identifier=self.config.dataset_identifier,
465 reference_package=self.config.reference_package,
466 timestamp_version=self.config.timestamp_version,
467 metricNamePrefix=prefixName,
468 )
469 # copy plot info to be sure each action sees its own copy
470 plotInfo = kwargs.get("plotInfo")
471 plotKey = f"{self.config.connections.outputName}_{{name}}"
472 weakrefArgs = []
473 for name, action in self.config.atools.items():
474 kwargs["plotInfo"] = deepcopy(plotInfo)
475 try:
476 actionResult = action(data, **kwargs)
477 except Exception:
478 tb = traceback.format_exc()
479 caughtErrors[name] = tb
481 continue
482 metricAccumulate = []
483 for resultName, value in actionResult.items():
484 match value:
485 case PlotTypes():
486 setattr(results, plotKey.format(name=resultName), value)
487 weakrefArgs.append(value)
488 case Measurement():
489 metricAccumulate.append(value)
490 # only add the metrics if there are some
491 if metricAccumulate:
492 results.metrics[name] = metricAccumulate
493 # Wrap the return struct in a finalizer so that when results is
494 # garbage collected the plots will be closed.
495 # TODO: This finalize step closes all open plots at the conclusion
496 # of a task. When DM-39114 is implemented, this step should not
497 # be required and may be removed.
498 weakref.finalize(results, _plotCloser, *weakrefArgs)
499 results.__caughtErrors = caughtErrors
500 return results
502 def run(self, *, data: KeyedData | None = None, **kwargs) -> Struct:
503 """Produce the outputs associated with this `PipelineTask`.
505 Parameters
506 ----------
507 data : `KeyedData`
508 The input data from which all `AnalysisTools` will run and produce
509 outputs. A side note, the python typing specifies that this can be
510 None, but this is only due to a limitation in python where in order
511 to specify that all arguments be passed only as keywords the
512 argument must be given a default. This argument most not actually
513 be None.
514 **kwargs
515 Additional arguments that are passed through to the `AnalysisTools`
516 specified in the configuration.
518 Returns
519 -------
520 results : `~lsst.pipe.base.Struct`
521 The accumulated results of all the plots and metrics produced by
522 this `PipelineTask`.
524 Raises
525 ------
526 ValueError
527 Raised if the supplied data argument is `None`
528 """
529 if data is None:
530 raise ValueError("data must not be none")
531 if "bands" not in kwargs:
532 kwargs["bands"] = list(self.config.bands)
533 if "plotInfo" not in kwargs:
534 kwargs["plotInfo"] = _StandinPlotInfo()
535 kwargs["plotInfo"]["bands"] = kwargs["bands"]
536 return self._runTools(data, **kwargs)
538 def runQuantum(
539 self,
540 butlerQC: QuantumContext,
541 inputRefs: InputQuantizedConnection,
542 outputRefs: OutputQuantizedConnection,
543 ) -> None:
544 """Override default runQuantum to load the minimal columns necessary
545 to complete the action.
547 Parameters
548 ----------
549 butlerQC : `~lsst.pipe.base.QuantumContext`
550 A butler which is specialized to operate in the context of a
551 `lsst.daf.butler.Quantum`.
552 inputRefs : `InputQuantizedConnection`
553 Datastructure whose attribute names are the names that identify
554 connections defined in corresponding `PipelineTaskConnections`
555 class. The values of these attributes are the
556 `lsst.daf.butler.DatasetRef` objects associated with the defined
557 input/prerequisite connections.
558 outputRefs : `OutputQuantizedConnection`
559 Datastructure whose attribute names are the names that identify
560 connections defined in corresponding `PipelineTaskConnections`
561 class. The values of these attributes are the
562 `lsst.daf.butler.DatasetRef` objects associated with the defined
563 output connections.
564 """
565 inputs = butlerQC.get(inputRefs)
566 dataId = butlerQC.quantum.dataId
567 plotInfo = self.parsePlotInfo(inputs, dataId)
568 # We implicitly assume that 'data' has been defined, but do not have a
569 # corresponding input connection in the base class. Thus, we capture
570 # and re-raise the error with a more helpful message.
571 try:
572 # data has to be popped out to avoid duplication in the call to the
573 # `run` method.
574 inputData = inputs.pop("data")
575 except KeyError:
576 raise RuntimeError("'data' is a required input connection, but is not defined.")
577 data = self.loadData(inputData)
578 outputs = self.run(data=data, plotInfo=plotInfo, **inputs)
579 self.putByBand(butlerQC, outputs, outputRefs)
581 def putByBand(self, butlerQC: QuantumContext, outputs: Struct, outputRefs: OutputQuantizedConnection):
582 """Handle the outputs by band.
584 This is a convenience method to handle the case where the
585 PipelineTaskConnection had to instantiate multiple output connections
586 for plots to loop over bands.
588 Parameters
589 ----------
590 butlerQC : `~lsst.pipe.base.QuantumContext`
591 A butler which is specialized to operate in the context of a
592 `lsst.daf.butler.Quantum`.
593 outputs : `~lsst.pipe.base.Struct`
594 The accumulated results of all the plots and metrics produced by
595 the `run` method of this `PipelineTask`.
596 outputRefs : `OutputQuantizedConnection`
597 Datastructure whose attribute names are the names that identify
598 connections defined in corresponding `PipelineTaskConnections`
599 class. The values of these attributes are the
600 `lsst.daf.butler.DatasetRef` objects associated with the defined
601 output connections.
602 """
603 for outputRefName in outputRefs.keys():
604 if outputRefName == "metrics":
605 butlerQC.put(getattr(outputs, outputRefName), getattr(outputRefs, outputRefName))
606 continue
608 name = outputRefName.split(self.config.connections.outputName, maxsplit=1)[1]
610 datasetRef = getattr(outputRefs, outputRefName)
611 if hasattr(datasetRef, "__iter__"):
612 for outputRef in datasetRef:
613 band = outputRef.dataId.get("band", "")
614 # name would already have a leading underscore.
615 newOutputName = f"{self.config.connections.outputName}_{band}{name}"
616 if dataset := getattr(outputs, newOutputName, None):
617 butlerQC.put(dataset, outputRef)
618 else:
619 if dataset := getattr(outputs, outputRefName, None):
620 butlerQC.put(dataset, datasetRef)
622 # If any of the actions encountered errors, raise an
623 # AnnotatedPartialOutputsError with the details.
624 if outputs.__caughtErrors:
625 e = PartialAtoolsFailureError(outputs.__caughtErrors)
626 error = AnnotatedPartialOutputsError.annotate(e, self, log=self.log)
627 raise error from e
629 def _populatePlotInfoWithDataId(
630 self, plotInfo: MutableMapping[str, Any], dataId: DataCoordinate | None
631 ) -> None:
632 """Update the plotInfo with the dataId values.
634 Parameters
635 ----------
636 plotInfo : `dict`
637 The plotInfo dictionary to update.
638 dataId : `lsst.daf.butler.DataCoordinate`
639 The dataId to use to update the plotInfo.
640 """
641 if dataId is not None: 641 ↛ exitline 641 didn't return from function '_populatePlotInfoWithDataId' because the condition on line 641 was always true
642 plotInfo.update(dataId.mapping)
644 def parsePlotInfo(
645 self, inputs: Mapping[str, Any] | None, dataId: DataCoordinate | None, connectionName: str = "data"
646 ) -> Mapping[str, str]:
647 """Parse the inputs and dataId to get the information needed to
648 to add to the figure.
650 Parameters
651 ----------
652 inputs: `dict`
653 The inputs to the task
654 dataCoordinate: `lsst.daf.butler.DataCoordinate`
655 The dataId that the task is being run on.
656 connectionName: `str`, optional
657 Name of the input connection to use for determining table name.
659 Returns
660 -------
661 plotInfo : `dict`
662 """
664 if inputs is None:
665 tableName = ""
666 run = ""
667 else:
668 tableName = inputs[connectionName].ref.datasetType.name
669 run = inputs[connectionName].ref.run
671 # Initialize the plot info dictionary
672 plotInfo = {"tableName": tableName, "run": run}
674 self._populatePlotInfoWithDataId(plotInfo, dataId)
675 return plotInfo
677 def loadData(self, handle: DeferredDatasetHandle, names: Iterable[str] | None = None) -> KeyedData:
678 """Load the minimal set of keyed data from the input dataset.
680 Parameters
681 ----------
682 handle : `DeferredDatasetHandle`
683 Handle to load the dataset with only the specified columns.
684 names : `Iterable` of `str`
685 The names of keys to extract from the dataset.
686 If `names` is `None` then the `collectInputNames` method
687 is called to generate the names.
688 For most purposes these are the names of columns to load from
689 a catalog or data frame.
691 Returns
692 -------
693 result: `KeyedData`
694 The dataset with only the specified keys loaded.
695 """
696 if names is None:
697 names = self.collectInputNames()
698 return cast(KeyedData, handle.get(parameters={"columns": names}))
700 def collectInputNames(self) -> Iterable[str]:
701 """Get the names of the inputs.
703 If using the default `loadData` method this will gather the names
704 of the keys to be loaded from an input dataset.
706 Returns
707 -------
708 inputs : `Iterable` of `str`
709 The names of the keys in the `KeyedData` object to extract.
711 """
712 inputs = set()
714 if not (localBands := self.config.bands):
715 localBands = [""]
716 for band in localBands:
717 for action in self.config.atools:
718 for key, _ in action.getFormattedInputSchema(band=band):
719 inputs.add(key)
720 return inputs