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