Coverage for python/lsst/analysis/tools/tasks/metadataAnalysis.py: 40%
118 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-20 10:14 +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__ = (
24 "MetadataAnalysisConfig",
25 "DatasetMetadataAnalysisTask",
26 "TaskMetadataAnalysisTask",
27 "AggregatedTaskMetadataAnalysisConfig",
28 "AggregatedTaskMetadataAnalysisTask",
29)
31from lsst.pex.config import Field, ListField
32from lsst.pipe.base import UpstreamFailureNoWorkFound, connectionTypes
34from ..interfaces import AnalysisBaseConfig, AnalysisBaseConnections, AnalysisPipelineTask
37class MetadataAnalysisConnections(
38 AnalysisBaseConnections,
39 dimensions={},
40 defaultTemplates={"inputName": "", "outputName": "", "datasetType": "", "storageClass": ""},
41):
42 def __init__(self, *, config=None):
43 """Customize the connections for a specific task's or dataset's
44 metadata.
46 Parameters
47 ----------
48 config : `MetadataAnalysisConfig`
49 A config for analyzing task or dataset metadata with this
50 connection.
51 """
52 # The following must come before super().__init__ so the input
53 # dimensions are propagated into the output metric bundle by
54 # the base class __init__
55 self.dimensions = frozenset(config.inputDimensions)
56 super().__init__(config=config)
58 self.data = connectionTypes.Input(
59 doc="Input dataset to extract metadata from.",
60 name=config.connections.inputName,
61 storageClass=config.connections.storageClass,
62 deferLoad=True,
63 dimensions=self.dimensions,
64 )
67class MetadataAnalysisConfig(
68 AnalysisBaseConfig,
69 pipelineConnections=MetadataAnalysisConnections,
70):
71 inputDimensions = ListField(
72 # Sort to ensure default order is consistent between runs
73 default=sorted(MetadataAnalysisConnections.dimensions),
74 dtype=str,
75 doc="The dimensions of the input dataset.",
76 )
77 raiseNoWorkFoundOnEmptyMetadata = Field(
78 dtype=bool,
79 default=False,
80 doc="Raise a NoWorkFound error if none of the configured metrics are in the task metadata.",
81 )
82 raiseNoWorkFoundOnIncompleteMetadata = Field(
83 dtype=bool,
84 default=False,
85 doc="Raise NoWorkFound if any of the configured metrics are not in the task metadata.",
86 )
88 def validate(self):
89 super().validate()
90 if self.raiseNoWorkFoundOnEmptyMetadata and self.raiseNoWorkFoundOnIncompleteMetadata:
91 raise ValueError("At most one 'raiseNoWorkFound' option may be True at a time.")
94class DatasetMetadataAnalysisConfig(MetadataAnalysisConfig):
96 def setDefaults(self):
97 super().setDefaults()
98 self.raiseNoWorkFoundOnIncompleteMetadata = True
101class DatasetMetadataAnalysisTask(AnalysisPipelineTask):
102 ConfigClass = DatasetMetadataAnalysisConfig
103 _DefaultName = "datasetMetadataAnalysis"
105 def runQuantum(self, butlerQC, inputRefs, outputRefs):
106 dataId = butlerQC.quantum.dataId
107 inputs = butlerQC.get(inputRefs)
108 plotInfo = self.parsePlotInfo(inputs, dataId)
109 data = inputs["data"].get()
111 # Collect all the metrics that are configured to run.
112 metadata = {}
113 for name in self.config.atools.fieldNames:
114 atool = getattr(self.config.atools, name)
115 if hasattr(atool, "metrics"):
116 for metric in atool.metrics:
117 # Check if the metric uses base key prefixes.
118 if atool.metricsPrefixedWithBaseKeys.get(metric):
119 # Find all keys prefixed with the base key and collect
120 # them in a dictionary.
121 metadata[metric] = {k: v for k, v in data.metadata.items() if k.startswith(metric)}
122 else:
123 # Retrieve the metric directly if it's not prefixed.
124 metadata[metric] = data.metadata.get(metric)
125 # Check if the retrieved metadata is empty.
126 if not metadata[metric] and self.config.raiseNoWorkFoundOnIncompleteMetadata:
127 raise UpstreamFailureNoWorkFound(
128 f"Metadata entry '{metric}' is empty for {inputRefs.data.datasetType.name}, "
129 f"or it is not one of {data.metadata.getOrderedNames()}."
130 )
131 if self.config.raiseNoWorkFoundOnEmptyMetadata and not any(v for v in metadata.values()):
132 raise UpstreamFailureNoWorkFound(
133 "All configured metadata entries are missing from "
134 f"{inputRefs.data.datasetType.name}."
135 )
137 outputs = self.run(data={"metadata_metrics": metadata}, plotInfo=plotInfo)
138 butlerQC.put(outputs, outputRefs)
141class TaskMetadataAnalysisTask(AnalysisPipelineTask):
142 ConfigClass = MetadataAnalysisConfig
143 _DefaultName = "taskMetadataAnalysis"
145 def runQuantum(self, butlerQC, inputRefs, outputRefs):
146 dataId = butlerQC.quantum.dataId
147 inputs = butlerQC.get(inputRefs)
148 plotInfo = self.parsePlotInfo(inputs, dataId)
149 metadata = inputs["data"].get().to_dict()
150 taskName = inputRefs.data.datasetType.name
151 taskName = taskName[: taskName.find("_")]
152 if not metadata:
153 raise UpstreamFailureNoWorkFound(f"No metadata entries for {taskName}.")
154 if self.config.raiseNoWorkFoundOnEmptyMetadata or self.config.raiseNoWorkFoundOnIncompleteMetadata:
155 self.validateMetrics(metadata, taskName)
156 outputs = self.run(data=metadata, plotInfo=plotInfo)
157 butlerQC.put(outputs, outputRefs)
159 def validateMetrics(self, metadata, taskName):
160 """Raise NoWorkFound if there are insufficent metrics in the task
161 metadata.
163 Parameters
164 ----------
165 metadata : `dict`
166 The task metadata converted to a dict.
167 taskName : `str`
168 The name of the task to extract metadata from
170 Raises
171 ------
172 NoWorkFound
173 If none of the metrics are in the metadata.
174 """
175 for fieldName in self.config.atools.fieldNames:
176 atool = getattr(self.config.atools, fieldName)
177 subTaskNames = getattr(atool, "subTaskNames", None) or {}
178 for key in atool.metrics.keys():
179 if key in metadata[taskName].keys():
180 if self.config.raiseNoWorkFoundOnEmptyMetadata:
181 return
182 elif subtaskName := subTaskNames.get(key):
183 if f"{taskName}:{subtaskName}" not in metadata:
184 raise UpstreamFailureNoWorkFound(
185 f"Subtask {subtaskName!r} was not found in {taskName} metadata"
186 )
187 if key in metadata[f"{taskName}:{subtaskName}"]:
188 if self.config.raiseNoWorkFoundOnEmptyMetadata:
189 return
190 else:
191 if self.config.raiseNoWorkFoundOnIncompleteMetadata:
192 raise UpstreamFailureNoWorkFound(
193 f"Metric {key!r} was not found in {taskName} metadata"
194 )
195 if self.config.raiseNoWorkFoundOnEmptyMetadata:
196 raise UpstreamFailureNoWorkFound(
197 f"None of the specified metrics were found in the {taskName} metadata"
198 )
201class AggregatedTaskMetadataAnalysisConnections(
202 AnalysisBaseConnections,
203 dimensions={},
204 defaultTemplates={"inputName": "", "outputName": "", "storageClass": "TaskMetadata"},
205):
206 def __init__(self, *, config=None):
207 """Allow for the dimensions of the input and output data and task
208 to be changed from the defaults via the corresponding config parameter.
209 This enables the task to work with any TaskMetadata.
211 Parameters
212 ----------
213 config : `AggregatedTaskMetadataAnalysisConfig`
214 Configuration for this task.
215 """
216 super().__init__(config=config)
218 self.data = connectionTypes.Input(
219 doc="Task metadata to aggregate across additional dimensions.",
220 name=config.connections.inputName,
221 storageClass=config.connections.storageClass,
222 deferLoad=True,
223 dimensions=frozenset(config.inputDataDimensions),
224 multiple=True,
225 )
227 self.dimensions.update(frozenset(config.outputDimensions))
230class AggregatedTaskMetadataAnalysisConfig(
231 AnalysisBaseConfig,
232 pipelineConnections=AggregatedTaskMetadataAnalysisConnections,
233):
234 inputDataDimensions = ListField(
235 default=["instrument", "visit", "detector"],
236 dtype=str,
237 doc="Dimensions of the input task metadata datasets.",
238 )
239 outputDimensions = ListField(
240 default=["instrument", "visit"],
241 dtype=str,
242 doc="Dimensions of the output metric bundle. Also this task's dimensions.",
243 )
246class AggregatedTaskMetadataAnalysisTask(AnalysisPipelineTask):
247 ConfigClass = AggregatedTaskMetadataAnalysisConfig
248 _DefaultName = "aggregatedTaskMetadataAnalysis"
250 def _collectData(self, handles, taskName):
251 """Collect metric vectors from a list of deferred dataset handles.
253 Parameters
254 ----------
255 handles : `list`
256 Deferred dataset handles whose ``.get().to_dict()`` returns the
257 task metadata as a nested dict.
258 taskName : `str`
259 The name of the task, used as the top-level key in the metadata
260 dict (subtask keys are of the form ``taskName:subTaskName``).
262 Returns
263 -------
264 data : `dict` [`str`, `list` [`float`]]
265 Mapping from metric name to list of per-input values.
267 Raises
268 ------
269 lsst.pipe.base.UpstreamFailureNoWorkFound
270 If no data could be collected from any input.
271 """
272 data: dict[str, list[float]] = {}
273 for i, handle in enumerate(handles):
274 metadata = handle.get().to_dict()
275 if not metadata:
276 continue
277 for atool in self.config.atools:
278 if not hasattr(atool, "metrics"): 278 ↛ 279line 278 didn't jump to line 279 because the condition on line 278 was never true
279 continue
280 subTaskNames = getattr(atool, "subTaskNames", None) or {}
281 for metric in atool.metrics.keys():
282 if metric in subTaskNames:
283 taskFullName = f"{taskName}:{subTaskNames[metric]}"
284 else:
285 taskFullName = taskName
286 value = metadata.get(taskFullName, {}).get(metric)
287 if value is not None:
288 if metric not in data.keys():
289 data[metric] = [float("nan")] * len(handles)
290 data[metric][i] = float(value)
292 if not data:
293 raise UpstreamFailureNoWorkFound(f"No metadata entries found for {taskName}.")
295 return data
297 def runQuantum(self, butlerQC, inputRefs, outputRefs):
298 inputs = butlerQC.get(inputRefs)
300 # Collecting data from [taskName]_metedata datasetType.
301 # Extract taskName from the datasetType name.
302 taskName = inputRefs.data[0].datasetType.name
303 taskName = taskName[: taskName.find("_")]
305 data = self._collectData(inputs["data"], taskName)
306 outputs = self.run(data=data)
307 butlerQC.put(outputs, outputRefs)