Coverage for credoai/prism/compare.py: 94%

1"""Performs comparison between 2 pipelines""" 

2from typing import Literal, Optional 

3 

4from connect.evidence.containers import MetricContainer 

5 

6from credoai.evaluators.utils.validation import check_instance 

7from credoai.lens import Lens 

8from credoai.prism.comparators.metric_comparator import MetricComparator 

9from credoai.prism.task import Task 

10from credoai.utils import flatten_list 

11from credoai.utils.common import ValidationError 

12 

13 

14class Compare(Task): 

15 """ 

16 Compare results across multiple pipelines. 

17 

18 This class coordinates prism.comparators objects in order to execute 

19 comparisons across an arbitrary amount of Lens objects. 

20 

21 The code execute the following steps: 

22 

23 1. Extract results from the Lens objects 

24 2. Call the suitable comparators depending on the results container type 

25 3. Aggregate results from multiple comparators (currently only metrics supported) 

26 

27 Parameters 

28 ---------- 

29 ref_type: str 

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

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

32 ref : Optional[str], optional 

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

34 defined when instantiating Lens objects, by the usage of credo.artifacts. If None, the 

35 first in the list will be used as a reference, by default None. 

36 operation : Literal["diff", "ratio", "perc", "perc_diff"], optional 

37 Indicates which operation is computed during the comparison. The accepted 

38 options are: 

39 

40 "diff": x - ref, 

41 "ratio": x / ref, 

42 "perc": x * 100 / ref, 

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

44 

45 abs : bool, optional 

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

47 """ 

48 

49 SUPPORTED_CONTAINERS = [MetricContainer] 

50 

51 def __init__( 

52 self, 

53 ref_type: str = "model", 

54 ref: Optional[str] = None, 

55 operation: Literal["diff", "ratio", "perc", "perc_diff"] = "diff", 

56 abs: bool = False, 

57 ): 

58 self.ref = ref 

59 self.ref_type = ref_type 

60 self.operation = operation 

61 self.abs = abs 

62 super().__init__() 

63 

64 def _validate(self): 

65 """ 

66 Validate that parameters are in the correct format. 

67 """ 

68 for evaluator in self.pipelines: 

69 check_instance(evaluator, Lens) 

70 if len(self.pipelines) < 2: 

71 raise ValidationError("At least 2 lens objects are needed for a comparison") 

72 

73 def _setup(self): 

74 pipesteps = flatten_list([x.pipeline for x in self.pipelines]) 

75 # Propagate step identifier to results 

76 for step in pipesteps: 

77 for result in step.evaluator.results: 

78 # Create the id property for each of the containers taking the Step identifier 

79 result.id = step.id 

80 self.containers = flatten_list([x.evaluator.results for x in pipesteps]) 

81 # Remove unsupported containers 

82 self.supported_results = [ 

83 x for x in self.containers if type(x) in self.SUPPORTED_CONTAINERS 

84 ] 

85 # Get default reference value if non is provided 

86 if not self.ref: 

87 self.ref = self.pipelines[0].__dict__[self.ref_type].name 

88 # TODO: LOG this -> (f"Reference {self.ref_type}: {self.ref}") 

89 

90 def run(self): 

91 """ 

92 Runs the comparisons. 

93 """ 

94 # TODO: Add a splitter for different type of containers 

95 # When we have other comparators we can use the suitable ones depending 

96 # on container type. Potentially also dependent on evaluator 

97 self.results = MetricComparator( 

98 self.supported_results, self.ref_type, self.ref, self.operation, self.abs 

99 ).compare() 

100 return self 

101 

102 def get_results(self): 

103 """ 

104 Returns the comparison results 

105 """ 

106 return self.results.comparisons