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 

22 

23"""Base class implementation for the classes needed in creating `PipelineTasks` 

24which execute `AnalysisTools`. 

25 

26The classes defined in this module have all the required behaviors for 

27defining, introspecting, and executing `AnalysisTools` against an input dataset 

28type. 

29 

30Subclasses of these tasks should specify specific datasets to consume in their 

31connection classes and should specify a unique name 

32""" 

33 

34__all__ = ("AnalysisBaseConnections", "AnalysisBaseConfig", "AnalysisPipelineTask") 

35 

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 

43 

44import matplotlib.pyplot as plt 

45from lsst.verify import Measurement 

46 

47if TYPE_CHECKING: 

48 from lsst.daf.butler import DatasetRef, DeferredDatasetHandle 

49 from lsst.pipe.base import QuantumContext 

50 

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 

66 

67from ._actions import JointAction, MetricAction, NoMetric 

68from ._analysisTools import AnalysisTool 

69from ._interfaces import KeyedData, PlotTypes 

70from ._metricMeasurementBundle import MetricMeasurementBundle 

71 

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}) 

76 

77 

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) 

85 

86 

87class PartialWorkFoundError(AlgorithmError): 

88 """Raised if an analysis tools action has partially failed. 

89 

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 """ 

96 

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}") 

100 

101 @property 

102 def metadata(self) -> dict: 

103 return self._caughtErrors 

104 

105 

106class AnalysisBaseConnections( 

107 PipelineTaskConnections, dimensions={}, defaultTemplates={"outputName": "Placeholder"} 

108): 

109 r"""Base class for Connections used for AnalysisTools PipelineTasks. 

110 

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``. 

114 

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. 

120 

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 """ 

125 

126 metrics = ct.Output( 

127 doc="Metrics calculated on input dataset type", 

128 name="{outputName}_metrics", 

129 storageClass="MetricMeasurementBundle", 

130 ) 

131 

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) 

142 

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 

148 

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 

162 

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") 

176 

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)) 

181 

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}) 

195 

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 ) 

204 

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 

223 

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) 

232 

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 

273 

274 

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 

300 

301 

302class AnalysisBaseConfig(PipelineTaskConfig, pipelineConnections=AnalysisBaseConnections): 

303 """Base class for all configs used to define an `AnalysisPipelineTask`. 

304 

305 This base class defines two fields that should be used in all subclasses, 

306 atools, and bands. 

307 

308 The ``atools`` field is where the user configures which analysis tools will 

309 be run as part of this `PipelineTask`. 

310 

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 """ 

315 

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 ) 

349 

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) 

386 

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() 

395 

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") 

401 

402 

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 """ 

407 

408 def __missing__(self, key): 

409 return "" 

410 

411 

412class AnalysisPipelineTask(PipelineTask): 

413 """Base class for `PipelineTasks` intended to run `AnalysisTools`. 

414 

415 The run method will run all of the `AnalysisTools` defined in the config 

416 class. 

417 """ 

418 

419 # Typing config because type checkers dont know about our Task magic 

420 config: AnalysisBaseConfig 

421 ConfigClass = AnalysisBaseConfig 

422 

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 ) 

435 

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 

481 

482 def run(self, *, data: KeyedData | None = None, **kwargs) -> Struct: 

483 """Produce the outputs associated with this `PipelineTask`. 

484 

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. 

497 

498 Returns 

499 ------- 

500 results : `~lsst.pipe.base.Struct` 

501 The accumulated results of all the plots and metrics produced by 

502 this `PipelineTask`. 

503 

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) 

517 

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. 

526 

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) 

560 

561 def putByBand(self, butlerQC: QuantumContext, outputs: Struct, outputRefs: OutputQuantizedConnection): 

562 """Handle the outputs by band. 

563 

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. 

567 

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 

587 

588 name = outputRefName.split(self.config.connections.outputName, maxsplit=1)[1] 

589 

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) 

601 

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 

608 

609 def _populatePlotInfoWithDataId( 

610 self, plotInfo: MutableMapping[str, Any], dataId: DataCoordinate | None 

611 ) -> None: 

612 """Update the plotInfo with the dataId values. 

613 

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) 

623 

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. 

629 

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. 

638 

639 Returns 

640 ------- 

641 plotInfo : `dict` 

642 """ 

643 

644 if inputs is None: 

645 tableName = "" 

646 run = "" 

647 else: 

648 tableName = inputs[connectionName].ref.datasetType.name 

649 run = inputs[connectionName].ref.run 

650 

651 # Initialize the plot info dictionary 

652 plotInfo = {"tableName": tableName, "run": run} 

653 

654 self._populatePlotInfoWithDataId(plotInfo, dataId) 

655 return plotInfo 

656 

657 def loadData(self, handle: DeferredDatasetHandle, names: Iterable[str] | None = None) -> KeyedData: 

658 """Load the minimal set of keyed data from the input dataset. 

659 

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. 

670 

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})) 

679 

680 def collectInputNames(self) -> Iterable[str]: 

681 """Get the names of the inputs. 

682 

683 If using the default `loadData` method this will gather the names 

684 of the keys to be loaded from an input dataset. 

685 

686 Returns 

687 ------- 

688 inputs : `Iterable` of `str` 

689 The names of the keys in the `KeyedData` object to extract. 

690 

691 """ 

692 inputs = set() 

693 

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