Coverage for python/lsst/analysis/tools/atools/metadataMetrics.py: 65%

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__all__ = ( 

24 "AggregatedTaskMetadataMetricTool", 

25 "DatasetMetadataMetricTool", 

26 "TaskMetadataMetricTool", 

27) 

28 

29 

30from lsst.pex.config import DictField, Field 

31 

32from ..actions.keyedData import KeyedDataKeyAccessAction 

33from ..interfaces import AnalysisTool 

34 

35 

36class MetadataMetricTool(AnalysisTool): 

37 """Base class for tools designed to extract values from metadata""" 

38 

39 parameterizedBand = Field[bool]( 

40 doc="Does this MetadataMetricTool support band as a name parameter?", default=False 

41 ) 

42 

43 metrics = DictField[str, str](doc="The metrics to extract from the metadata and their respective units.") 

44 

45 newNames = DictField[str, str]( 

46 doc="New names to allocate to the extracted metrics. Keys are the current " 

47 "names, values are the new names.", 

48 default=None, 

49 optional=True, 

50 ) 

51 

52 @staticmethod 

53 def makeValidAttributeName(name): 

54 """Make a valid attribute name using a simple replacement.""" 

55 return name.replace(" ", "_") 

56 

57 def validate(self): 

58 for metric in self.metrics.keys(): 

59 if not self.makeValidAttributeName(metric).isidentifier(): 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

60 raise ValueError( 

61 f"{metric=} must be a valid identifier after replacing spaces with underscores." 

62 ) 

63 

64 

65class DatasetMetadataMetricTool(MetadataMetricTool): 

66 """Tool designed to extract values from metadata of data products""" 

67 

68 metricsPrefixedWithBaseKeys = DictField[str, bool]( 

69 doc="Whether metrics are prefixed with base keys. For each key (a metric name or base key), " 

70 "the corresponding boolean value specifies if the metric should be extracted using the base " 

71 "key as a prefix.", 

72 default={}, 

73 optional=True, 

74 ) 

75 

76 def finalize(self): 

77 for metric in self.metrics.keys(): 

78 setattr( 

79 self.process.filterActions, 

80 self.makeValidAttributeName(metric), 

81 KeyedDataKeyAccessAction(topLevelKey="metadata_metrics"), 

82 ) 

83 self.produce.metric.units = dict(self.metrics.items()) 

84 

85 if self.newNames is not None: 

86 self.produce.metric.newNames = dict(self.newNames.items()) 

87 

88 

89class TaskMetadataMetricTool(MetadataMetricTool): 

90 """This tool is designed to extract values from task metadata""" 

91 

92 taskName = Field[str]( 

93 doc="The name of the task to extract metadata from.", 

94 default=None, 

95 ) 

96 

97 subTaskNames = DictField[str, str]( 

98 doc="The names of the subtasks to extract metadata from. " 

99 "If the metric name is identified as one of the keys, then " 

100 "the corresponding value is taken as the subTask metadata " 

101 "from which to extract metadata.", 

102 default=None, 

103 optional=True, 

104 ) 

105 

106 def finalize(self): 

107 for metric in self.metrics.keys(): 

108 if self.subTaskNames is not None and metric in self.subTaskNames: 

109 taskFullName = f"{self.taskName}:{self.subTaskNames[metric]}" 

110 else: 

111 taskFullName = self.taskName 

112 setattr( 

113 self.process.filterActions, 

114 self.makeValidAttributeName(metric), 

115 KeyedDataKeyAccessAction(topLevelKey=taskFullName), 

116 ) 

117 self.produce.metric.units = dict(self.metrics.items()) 

118 

119 if self.newNames is not None: 

120 self.produce.metric.newNames = dict(self.newNames.items()) 

121 

122 

123class AggregatedTaskMetadataMetricTool(MetadataMetricTool): 

124 """Tool to compute aggregate statistics from task metadata across 

125 multiple inputs. 

126 """ 

127 

128 taskName = Field[str]( 

129 doc="The name of the task to extract metadata from.", 

130 default=None, 

131 ) 

132 

133 subTaskNames = DictField[str, str]( 

134 doc="The names of subtasks to extract metadata from. " 

135 "If the metric name is identified as one of the keys, then " 

136 "the corresponding value is taken as the subTask metadata " 

137 "from which to extract metadata.", 

138 default=None, 

139 optional=True, 

140 ) 

141 

142 aggregationUnits = DictField[str, str]( 

143 doc="Fixed units for specific aggregations, keyed by aggregation name " 

144 "(the suffix after the metric name, e.g. 'ct' or 'frac'). " 

145 "Overrides the source metric's unit for that aggregation.", 

146 default={}, 

147 optional=True, 

148 ) 

149 

150 def finalize(self): 

151 # Attribute names can't contain spaces, but metadata metric names 

152 # might. Remove any spaces from names. 

153 valid_metric_names = {self.makeValidAttributeName(k): v for k, v in self.metrics.items()} 

154 units = {} 

155 # This is the same as looping over the metric names, but it also 

156 # provides the aggregator name at the same time which is needed 

157 # for post-aggregation unit lookup. 

158 for name, action in self.process.calculateActions.items(): 

159 vector_key = getattr(action, "vectorKey", None) 

160 if vector_key is not None: 160 ↛ 158line 160 didn't jump to line 158 because the condition on line 160 was always true

161 valid_vector_key = self.makeValidAttributeName(vector_key) 

162 if valid_vector_key not in valid_metric_names: 

163 raise ValueError( 

164 f"Action {name!r} has vectorKey {vector_key!r} which is " 

165 f"not in metrics: {list(self.metrics.keys())}" 

166 ) 

167 agg_name = name[len(valid_vector_key) + 1 :] 

168 if self.aggregationUnits and agg_name in self.aggregationUnits: 

169 units[name] = self.aggregationUnits[agg_name] 

170 else: 

171 units[name] = valid_metric_names[valid_vector_key] 

172 self.produce.metric.units = units 

173 if self.newNames is not None: 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true

174 self.produce.metric.newNames = dict(self.newNames.items())