Coverage for sparkle/configurator/configurator.py: 88%

1#!/usr/bin/env python3 

2# -*- coding: UTF-8 -*- 

3"""Configurator class to use different algorithm configurators like SMAC.""" 

4from __future__ import annotations 

5from abc import abstractmethod 

6from typing import Callable 

7import ast 

8from statistics import mean 

9import operator 

10from pathlib import Path 

11 

12from runrunner import Runner, Run 

13from sparkle.solver import Solver 

14from sparkle.solver.validator import Validator 

15from sparkle.instance import InstanceSet 

16from sparkle.types import SparkleObjective 

17 

18 

19class Configurator: 

20 """Abstact class to use different configurators like SMAC.""" 

21 configurator_cli_path = Path(__file__).parent.resolve() / "configurator_cli.py" 

22 

23 def __init__(self: Configurator, validator: Validator, output_path: Path, 

24 base_dir: Path, tmp_path: Path, 

25 multi_objective_support: bool = False) -> None: 

26 """Initialize Configurator. 

27 

28 Args: 

29 validator: Validator object to validate configurations runs 

30 output_path: Output directory of the Configurator. 

31 objectives: The list of Sparkle Objectives the configurator has to 

32 optimize. 

33 base_dir: Where to execute the configuration 

34 tmp_path: Path for the temporary files of the configurator, optional 

35 multi_objective_support: Whether the configurator supports 

36 multi objective optimization for solvers. 

37 """ 

38 self.validator = validator 

39 self.output_path = output_path 

40 self.base_dir = base_dir 

41 self.tmp_path = tmp_path 

42 self.multiobjective = multi_objective_support 

43 self.scenario = None 

44 

45 @property 

46 def scenario_class(self: Configurator) -> ConfigurationScenario: 

47 """Return the scenario class of the configurator.""" 

48 return ConfigurationScenario 

49 

50 @abstractmethod 

51 def configure(self: Configurator, 

52 scenario: ConfigurationScenario, 

53 validate_after: bool = True, 

54 sbatch_options: list[str] = [], 

55 num_parallel_jobs: int = None, 

56 base_dir: Path = None, 

57 run_on: Runner = Runner.SLURM) -> Run: 

58 """Start configuration job. 

59 

60 Args: 

61 scenario: ConfigurationScenario to execute. 

62 validate_after: Whether to validate the configuration on the training set 

63 afterwards or not. 

64 sbatch_options: List of slurm batch options to use 

65 num_parallel_jobs: The maximum number of jobs to run in parallel 

66 base_dir: The base_dir of RunRunner where the sbatch scripts will be placed 

67 run_on: On which platform to run the jobs. Default: Slurm. 

68 

69 Returns: 

70 A RunRunner Run object. 

71 """ 

72 raise NotImplementedError 

73 

74 def get_optimal_configuration( 

75 self: Configurator, 

76 scenario: ConfigurationScenario, 

77 aggregate_config: Callable = mean) -> tuple[float, str]: 

78 """Returns optimal value and configuration string of solver on instance set.""" 

79 self.validator.out_dir = scenario.validation 

80 results = self.validator.get_validation_results( 

81 scenario.solver, 

82 scenario.instance_set, 

83 source_dir=scenario.validation, 

84 subdir=Path()) 

85 # Group the results per configuration 

86 objective = scenario.sparkle_objective 

87 value_column = results[0].index(objective.name) 

88 config_column = results[0].index("Configuration") 

89 configurations = list(set(row[config_column] for row in results[1:])) 

90 config_scores = [] 

91 for config in configurations: 

92 values = [float(row[value_column]) 

93 for row in results[1:] if row[1] == config] 

94 config_scores.append(aggregate_config(values)) 

95 

96 comparison = operator.lt if objective.minimise else operator.gt 

97 

98 # Return optimal value 

99 min_index = 0 

100 current_optimal = config_scores[min_index] 

101 for i, score in enumerate(config_scores): 

102 if comparison(score, current_optimal): 

103 min_index, current_optimal = i, score 

104 

105 # Return the optimal configuration dictionary as commandline args 

106 config_str = configurations[min_index].strip(" ") 

107 if config_str.startswith("{"): 

108 config = ast.literal_eval(config_str) 

109 config_str = " ".join([f"-{key} '{config[key]}'" for key in config]) 

110 return current_optimal, config_str 

111 

112 @staticmethod 

113 def organise_output(output_source: Path, output_target: Path) -> None | str: 

114 """Method to restructure and clean up after a single configurator call.""" 

115 raise NotImplementedError 

116 

117 def set_scenario_dirs(self: Configurator, 

118 solver: Solver, instance_set: InstanceSet) -> None: 

119 """Patching method to allow the rebuilding of configuration scenario.""" 

120 raise NotImplementedError 

121 

122 def get_status_from_logs(self: Configurator) -> None: 

123 """Method to scan the log files of the configurator for warnings.""" 

124 raise NotImplementedError 

125 

126 

127class ConfigurationScenario: 

128 """Template class to handle a configuration scenarios.""" 

129 def __init__(self: ConfigurationScenario, 

130 solver: Solver, 

131 instance_set: InstanceSet, 

132 sparkle_objectives: list[SparkleObjective], 

133 parent_directory: Path)\ 

134 -> None: 

135 """Initialize scenario paths and names. 

136 

137 Args: 

138 solver: Solver that should be configured. 

139 instance_set: Instances object for the scenario. 

140 sparkle_objectives: Sparkle Objectives to optimize. 

141 parent_directory: Directory in which the scenario should be placed. 

142 """ 

143 self.solver = solver 

144 self.instance_set = instance_set 

145 self.sparkle_objectives = sparkle_objectives 

146 self.name = f"{self.solver.name}_{self.instance_set.name}" 

147 

148 self.directory = parent_directory / self.name 

149 self.scenario_file_path = self.directory / f"{self.name}_scenario.txt" 

150 self.validation: Path = None 

151 self.tmp: Path = None 

152 self.results_directory: Path = None 

153 

154 def create_scenario(self: ConfigurationScenario, parent_directory: Path) -> None: 

155 """Create scenario with solver and instances in the parent directory. 

156 

157 This prepares all the necessary subdirectories related to configuration. 

158 

159 Args: 

160 parent_directory: Directory in which the scenario should be created. 

161 """ 

162 raise NotImplementedError 

163 

164 def create_scenario_file(self: ConfigurationScenario) -> Path: 

165 """Create a file with the configuration scenario. 

166 

167 Writes supplementary information to the target algorithm (algo =) as: 

168 algo = {configurator_target} {solver_directory} {sparkle_objective} 

169 """ 

170 raise NotImplementedError 

171 

172 def serialize(self: ConfigurationScenario) -> dict: 

173 """Serialize the configuration scenario.""" 

174 raise NotImplementedError 

175 

176 @classmethod 

177 def find_scenario(cls: ConfigurationScenario, 

178 directory: Path, 

179 solver: Solver, 

180 instance_set: InstanceSet) -> ConfigurationScenario | None: 

181 """Resolve a scenario from a directory and Solver / Training set.""" 

182 scenario_name = f"{solver.name}_{instance_set.name}" 

183 path = directory / f"{scenario_name}" / f"{scenario_name}_scenario.txt" 

184 if not path.exists(): 

185 return None 

186 return cls.from_file(path) 

187 

188 @staticmethod 

189 def from_file(scenario_file: Path) -> ConfigurationScenario: 

190 """Reads scenario file and initalises ConfigurationScenario.""" 

191 raise NotImplementedError