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

1#!/usr/bin/env python3 

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

3 

4from __future__ import annotations 

5 

6from sparkle.platform import \ 

7 generate_report_for_configuration as sgrfch 

8from sparkle.solver import Solver 

9from sparkle.instance import InstanceSet 

10from sparkle.configurator.configurator import Configurator, ConfigurationScenario 

11from sparkle.solver.validator import Validator 

12from sparkle.platform.output.structures import ValidationResults, ConfigurationResults 

13from sparkle.types import SolverStatus 

14 

15import json 

16from pathlib import Path 

17 

18 

19class ConfigurationOutput: 

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

21 

22 def __init__(self: ConfigurationOutput, path: Path, solver: Solver, 

23 configurator: Configurator, instance_set_train: InstanceSet, 

24 instance_set_test: InstanceSet, output: Path) -> None: 

25 """Initialize Configurator Output class. 

26 

27 Args: 

28 path: Path to configuration output directory 

29 solver: Solver object 

30 configurator: The configurator that was used 

31 instance_set_train: Instance set used for training 

32 instance_set_test: Instance set used for testing 

33 output: Path to the output directory 

34 """ 

35 self.solver = solver 

36 self.configurator = configurator 

37 self.instance_set_train = instance_set_train 

38 self.instance_set_test = instance_set_test 

39 self.directory = path 

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

41 

42 solver_dir_name = path.name 

43 scenario_file = path / f"{solver_dir_name}_scenario.txt" 

44 if not scenario_file.is_file(): 

45 raise Exception("Can't find scenario file") 

46 

47 # Sets scenario on configurator object 

48 self.configurator.scenario = \ 

49 configurator.scenario_class.from_file(scenario_file, self.solver, 

50 self.instance_set_train) 

51 self.configurator.scenario._set_paths(self.configurator.output_path) 

52 

53 # Retrieve all configurations 

54 config_path = path / "validation" / "configurations.csv" 

55 self.configurations = self.get_configurations(config_path) 

56 

57 # Retrieve best found configuration 

58 objective = self.configurator.scenario.sparkle_objective 

59 _, self.best_config = self.configurator.get_optimal_configuration( 

60 self.solver, self.instance_set_train, objective) 

61 

62 # Retrieves validation results for all configurations 

63 self.validation_results = [] 

64 for config in self.configurations: 

65 val_res = self.get_validation_data(self.instance_set_train, 

66 config) 

67 self.validation_results.append(val_res) 

68 

69 # Retrieve test validation results if they exist 

70 if self.instance_set_test is not None: 

71 self.validation_results_test = [] 

72 for config in self.configurations: 

73 val_res = self.get_validation_data(self.instance_set_test, 

74 config) 

75 self.validation_results_test.append(val_res) 

76 

77 def get_configurations(self: ConfigurationOutput, config_path: Path) -> list[dict]: 

78 """Read all configurations and transform them to dictionaries.""" 

79 configs = [] 

80 # Check if the path exists and is a file 

81 if config_path.exists() and config_path.is_file(): 

82 with config_path.open("r") as file: 

83 for line in file: 

84 config = Solver.config_str_to_dict(line.strip()) 

85 if config not in configs: 

86 configs.append(config) 

87 return configs 

88 

89 def get_validation_data(self: ConfigurationOutput, instance_set: InstanceSet, 

90 config: dict) -> ConfigurationResults: 

91 """Returns best config and ConfigurationResults for instance set.""" 

92 objective = self.configurator.scenario.sparkle_objective 

93 

94 # Retrieve found configuration 

95 _, best_config = self.configurator.get_optimal_configuration( 

96 self.solver, instance_set, objective) 

97 

98 # Retrieve validation results 

99 validator = Validator(self.directory) 

100 val_results = validator.get_validation_results( 

101 self.solver, instance_set, config=best_config, 

102 source_dir=self.directory, subdir="validation") 

103 header = val_results[0] 

104 results = [] 

105 value_column = header.index(objective.name) 

106 instance_column = header.index("Instance") 

107 status_column = header.index("Status") 

108 cpu_time_column = header.index("CPU Time") 

109 wall_time_column = header.index("Wallclock Time") 

110 for res in val_results[1:]: 

111 results.append([res[instance_column], SolverStatus(res[status_column]), 

112 res[value_column], res[cpu_time_column], 

113 res[wall_time_column]]) 

114 final_results = ValidationResults(self.solver, config, 

115 instance_set, results) 

116 perf_par = sgrfch.get_average_performance(val_results, 

117 objective) 

118 return ConfigurationResults(perf_par, 

119 final_results) 

120 

121 def serialize_configuration_results(self: ConfigurationOutput, 

122 cr: ConfigurationResults) -> dict: 

123 """Transform ConfigurationResults to dictionary format.""" 

124 return { 

125 "performance": cr.performance, 

126 "results": { 

127 "solver": cr.results.solver.name, 

128 "configuration": cr.results.configuration, 

129 "instance_set": cr.results.instance_set.name, 

130 "result_header": cr.results.result_header, 

131 "result_values": cr.results.result_vals, 

132 }, 

133 } 

134 

135 def serialize_scenario(self: ConfigurationOutput, 

136 scenario: ConfigurationScenario) -> dict: 

137 """Transform ConfigurationScenario to dictionary format.""" 

138 return { 

139 "number_of_runs": scenario.number_of_runs, 

140 "solver_calls": scenario.solver_calls, 

141 "cpu_time": scenario.cpu_time, 

142 "wallclock_time": scenario.wallclock_time, 

143 "cutoff_time": scenario.cutoff_time, 

144 "cutoff_length": scenario.cutoff_length, 

145 "sparkle_objective": scenario.sparkle_objective.name, 

146 "use_features": scenario.use_features, 

147 "configurator_target": scenario.configurator_target, 

148 "feature_data": scenario.feature_data, 

149 } 

150 

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

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

153 output_data = { 

154 "solver": self.solver.name if self.solver else None, 

155 "configurator": ( 

156 str(self.configurator.executable_path) if self.configurator else None 

157 ), 

158 "best_configuration": Solver.config_str_to_dict(self.best_config), 

159 "configurations": self.configurations, 

160 "scenario": self.serialize_scenario(self.configurator.scenario) 

161 if self.configurator.scenario else None, 

162 "training_results": [ 

163 self.serialize_configuration_results(validation_result) 

164 for validation_result in self.validation_results 

165 ], 

166 "test_set": ( 

167 [ 

168 self.serialize_configuration_results(validation_result) 

169 for validation_result in self.validation_results_test 

170 ] 

171 if self.instance_set_test else None 

172 ), 

173 } 

174 

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

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

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