Coverage for credoai/prism/comparators/metric_comparator.py: 95%

1"""Comparators for metric type containers""" 

2from typing import List 

3 

4from connect.evidence import MetricContainer 

5from pandas import DataFrame, concat 

6 

7from credoai.evaluators.utils.validation import check_instance 

8from credoai.prism.comparators.comparator import Comparator 

9 

10 

11class MetricComparator(Comparator): 

12 """ 

13 Class for comparing metric evidence objects. 

14 

15 Each metric is compared to the respective reference value. The reference value is the metric value 

16 associated to a specific model or dataset. The reference model/dataset are identified by the user, see 

17 ref type and ref in Parameters. 

18 

19 Supported comparisons for Metrics include differences, ratio, percentage ratio, and percentage 

20 difference. 

21 

22 Parameters 

23 ---------- 

24 EvidenceContainers : List[MetricContainer] 

25 A list of metric containers. 

26 ref_type: str 

27 Accepted values: model, assessment_data, training_data. Indicates which of the 

28 artifacts should be used as a refence, by default model. 

29 ref : str 

30 The model/dataset name by which to compare all others. Model/dataset names are 

31 defined when instantiating Lens objects, by the usage of credo.artifacts. 

32 operation : str 

33 Accepted operations: "diff", "ratio", "perc", "perc_diff", by default "diff" 

34 abs : bool, optional 

35 If true the absolute value of the operation is returned, by default False 

36 """ 

37 

38 OPERATIONS = { 

39 "diff": lambda x, ref: x - ref, 

40 "ratio": lambda x, ref: x / ref, 

41 "perc": lambda x, ref: x * 100 / ref, 

42 "perc_diff": lambda x, ref: ((x - ref) / ref) * 100, 

43 } 

44 ID_COLS = [ 

45 "evaluator", 

46 "model", 

47 "assessment_data", 

48 "training_data", 

49 "sensitive_feature", 

50 ] 

51 

52 def __init__( 

53 self, 

54 EvidenceContainers: List[MetricContainer], 

55 ref_type: str, 

56 ref: str, 

57 operation: str = "diff", 

58 abs: bool = False, 

59 ): 

60 # attributes all comparators will need 

61 self.ref_type = ref_type 

62 self.overall_ref = ref 

63 self.operation = self.OPERATIONS[operation] 

64 self.abs = abs 

65 super().__init__(EvidenceContainers) 

66 

67 def _setup(self): 

68 """Extracts all the results from the result containers""" 

69 self._extract_results_from_containers() 

70 

71 def _validate(self): 

72 """ 

73 Check that provided containers are all MetricContainer type 

74 """ 

75 for container in self.EvidenceContainers: 

76 check_instance(container, MetricContainer) 

77 

78 def compare(self): 

79 """ 

80 Runs all comparisons 

81 """ 

82 # Calculate scalar differences across all the metrics 

83 self.comparisons["scalar_comparison"] = self._scalar_operation() 

84 return self 

85 

86 def _extract_results_from_containers(self): 

87 """ 

88 Extract results from containers. 

89 """ 

90 # Create id columns from the result id. 

91 self.all_results = [ 

92 res.data.assign(id=res.id) for res in self.EvidenceContainers 

93 ] 

94 self.all_results = concat(self.all_results, ignore_index=True) 

95 self.all_results[self.ID_COLS] = self.all_results.id.str.split("~", expand=True) 

96 

97 # Drop columns with no variance 

98 self.all_results.drop("id", axis=1, inplace=True) 

99 

100 def _scalar_operation(self) -> DataFrame: 

101 """ 

102 Compares all each metric to a specific reference value 

103 

104 Returns 

105 ------- 

106 DataFrame 

107 Containing a comparison column with the results. The other columns 

108 include initial values and all other necessary identifiers. 

109 """ 

110 # Group by metrics and calculate comparison 

111 output = [] 

112 

113 # Remove reference type from ids, and add metric type to create unique groups 

114 to_grp_by = [x for x in self.ID_COLS if x != self.ref_type] + ["type"] 

115 

116 for _, results in self.all_results.groupby(to_grp_by): 

117 # Define reference value 

118 ref_value = results.value.loc[ 

119 results[self.ref_type] == self.overall_ref 

120 ].iloc[0] 

121 

122 if ref_value: 

123 results["comparison"] = self.operation(results.value, ref_value) 

124 else: 

125 results["comparison"] = None 

126 output.append(results) 

127 

128 final = concat(output, ignore_index=True) 

129 

130 if self.abs: 

131 final["comparison"] = final["comparison"].abs() 

132 

133 # Clean data frame 

134 nunique = final.nunique() 

135 cols_to_drop = nunique[nunique == 1].index 

136 final = final.drop(cols_to_drop, axis=1) 

137 final[final == "NA"] = None 

138 

139 return final