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

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

25 "TaskMetadataMetricTool", 

26) 

27 

28 

29from lsst.pex.config import DictField, Field 

30 

31from ..actions.keyedData import KeyedDataKeyAccessAction 

32from ..interfaces import AnalysisTool 

33 

34 

35class MetadataMetricTool(AnalysisTool): 

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

37 

38 parameterizedBand = Field[bool]( 

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

40 ) 

41 

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

43 

44 newNames = DictField[str, str]( 

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

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

47 default=None, 

48 optional=True, 

49 ) 

50 

51 @staticmethod 

52 def makeValidAttributeName(name): 

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

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

55 

56 def validate(self): 

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

58 if not self.makeValidAttributeName(metric).isidentifier(): 

59 raise ValueError( 

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

61 ) 

62 

63 

64class DatasetMetadataMetricTool(MetadataMetricTool): 

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

66 

67 metricsPrefixedWithBaseKeys = DictField[str, bool]( 

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

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

70 "key as a prefix.", 

71 default={}, 

72 optional=True, 

73 ) 

74 

75 def finalize(self): 

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

77 setattr( 

78 self.process.filterActions, 

79 self.makeValidAttributeName(metric), 

80 KeyedDataKeyAccessAction(topLevelKey="metadata_metrics"), 

81 ) 

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

83 

84 if self.newNames is not None: 

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

86 

87 

88class TaskMetadataMetricTool(MetadataMetricTool): 

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

90 

91 taskName = Field[str]( 

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

93 default=None, 

94 ) 

95 

96 subTaskNames = DictField[str, str]( 

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

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

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

100 "from which to extract metadata.", 

101 default=None, 

102 optional=True, 

103 ) 

104 

105 def finalize(self): 

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

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

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

109 else: 

110 taskFullName = self.taskName 

111 setattr( 

112 self.process.filterActions, 

113 self.makeValidAttributeName(metric), 

114 KeyedDataKeyAccessAction(topLevelKey=taskFullName), 

115 ) 

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

117 

118 if self.newNames is not None: 

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