Coverage for sparkle/platform/output/configuration_output.py: 92%

1"""Sparkle class to organise configuration output.""" 

2from __future__ import annotations 

3import ast 

4import json 

5from pathlib import Path 

6 

7from sparkle.solver import Solver 

8from sparkle.structures import PerformanceDataFrame 

9from sparkle.instance import InstanceSet, Instance_Set 

10from sparkle.configurator.configurator import Configurator, ConfigurationScenario 

11 

12 

13class ConfigurationOutput: 

14 """Class that collects configuration data and outputs it a JSON format.""" 

15 

16 def __init__(self: ConfigurationOutput, 

17 path: Path, 

18 configurator: Configurator, 

19 config_scenario: ConfigurationScenario, 

20 performance_data: PerformanceDataFrame, 

21 instance_set_test: InstanceSet, 

22 output: Path) -> None: 

23 """Initialize Configurator Output class. 

24 

25 Args: 

26 path: Path to configuration output directory 

27 configurator: The configurator that was used 

28 config_scenario: The scenario to output 

29 performance_data: Performance data 

30 instance_set_test: Instance set used for testing 

31 output: Path to the output directory 

32 """ 

33 self.solver = config_scenario.solver 

34 self.configurator = configurator 

35 self.instance_set_train = config_scenario.instance_set 

36 self.instance_set_test = instance_set_test 

37 self.directory = path 

38 self.config_scenario = config_scenario 

39 self.output = output / "configuration.json" if not output.is_file() else output 

40 

41 # Fix relative path 

42 if Path.cwd() in self.solver.directory.parents: 

43 self.solver = Solver( 

44 self.solver.directory.relative_to(Path.cwd()), 

45 ) 

46 if Path.cwd() in self.instance_set_train.directory.parents: 

47 self.instance_set_train = Instance_Set( 

48 self.instance_set_train.directory.relative_to(Path.cwd()) 

49 ) 

50 if (self.instance_set_test 

51 and Path.cwd() in self.instance_set_test.directory.parents): 

52 self.instance_set_test = Instance_Set( 

53 self.instance_set_test.directory.relative_to(Path.cwd()) 

54 ) 

55 # Retrieve all configurations 

56 solver_key = str(self.solver.directory) 

57 all_configurations = performance_data.get_value( 

58 solver_key, None, 

59 objective=self.config_scenario.sparkle_objective.name, 

60 solver_fields=[PerformanceDataFrame.column_configuration]) 

61 

62 # Turn in to dictionary and unique 

63 self.configurations_performances = [] 

64 for config in all_configurations: 

65 try: 

66 config = ast.literal_eval(config) 

67 if config not in self.configurations_performances: 

68 self.configurations_performances.append(config) 

69 except Exception: 

70 if isinstance(config, str): 

71 print("Failed to parse configuration:\n", config) 

72 # Retrieve configuration performances 

73 train_instances = [str(p) for p in self.instance_set_train.instance_paths] 

74 # Retrieve Default (No configuration) performance 

75 _, self.default_performance_train = performance_data.configuration_performance( 

76 solver_key, PerformanceDataFrame.missing_value, 

77 objective=self.config_scenario.sparkle_objective, 

78 instances=train_instances) 

79 

80 _, self.default_performance_per_instance_train =\ 

81 performance_data.configuration_performance( 

82 solver_key, PerformanceDataFrame.missing_value, 

83 objective=self.config_scenario.sparkle_objective, 

84 instances=train_instances, 

85 per_instance=True) 

86 

87 self.configurations_performances = [performance_data.configuration_performance( 

88 solver_key, config, 

89 objective=self.config_scenario.sparkle_objective, 

90 instances=train_instances) for config in self.configurations_performances] 

91 

92 # Retrieve best found configuration 

93 self.best_configuration, self.best_performance_train =\ 

94 performance_data.best_configuration( 

95 solver_key, 

96 objective=self.config_scenario.sparkle_objective, 

97 instances=train_instances) 

98 

99 # Retrieve best configuration per instance performances 

100 _, self.best_conf_performance_per_instance_train =\ 

101 performance_data.configuration_performance( 

102 solver_key, self.best_configuration, 

103 objective=self.config_scenario.sparkle_objective, 

104 instances=train_instances, 

105 per_instance=True) 

106 

107 if instance_set_test: 

108 test_instances = [str(p) for p in self.instance_set_test.instance_paths] 

109 # Retrieve default performance on the test set 

110 _, self.default_performance_test =\ 

111 performance_data.configuration_performance( 

112 solver_key, PerformanceDataFrame.missing_value, 

113 objective=self.config_scenario.sparkle_objective, 

114 instances=test_instances) 

115 _, self.default_performance_per_instance_test =\ 

116 performance_data.configuration_performance( 

117 solver_key, PerformanceDataFrame.missing_value, 

118 objective=self.config_scenario.sparkle_objective, 

119 instances=test_instances, 

120 per_instance=True) 

121 # Retrieve the best configuration test set performance 

122 _, self.best_performance_test = performance_data.configuration_performance( 

123 solver_key, self.best_configuration, 

124 objective=self.config_scenario.sparkle_objective, 

125 instances=test_instances, 

126 ) 

127 _, self.best_conf_performance_per_instance_test =\ 

128 performance_data.configuration_performance( 

129 solver_key, self.best_configuration, 

130 objective=self.config_scenario.sparkle_objective, 

131 instances=test_instances, 

132 per_instance=True) 

133 self.performance_data = performance_data 

134 

135 def write_output(self: ConfigurationOutput) -> None: 

136 """Write data into a JSON file.""" 

137 output_data = { 

138 "solver": self.solver.name, 

139 "configurator": self.configurator.name, 

140 "best_configuration": self.best_configuration, 

141 "best_performance_train": self.best_performance_train, 

142 "scenario": self.config_scenario.serialize() 

143 if self.configurator.scenario else None, 

144 "train_set_results": self.performance_data[self.performance_data.index.isin( 

145 [str(p) for p in self.instance_set_train.instance_paths], 

146 level=PerformanceDataFrame.index_instance)].to_json(), 

147 "test_set_results": (self.performance_data[self.performance_data.index.isin( 

148 [str(p) for p in self.instance_set_test.instance_paths], 

149 level=PerformanceDataFrame.index_instance)].to_json() 

150 if self.instance_set_test else None), 

151 } 

152 self.output.parent.mkdir(parents=True, exist_ok=True) 

153 with self.output.open("w") as f: 

154 json.dump(output_data, f, indent=4)