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

1#!/usr/bin/env python3 

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

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

4from __future__ import annotations 

5from pathlib import Path 

6 

7import runrunner as rrr 

8from runrunner import Runner, Run 

9 

10from sparkle.solver import Solver 

11from sparkle.instance import InstanceSet 

12from sparkle.structures import PerformanceDataFrame 

13from sparkle.types import SparkleObjective 

14 

15 

16class Configurator: 

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

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

19 

20 def __init__(self: Configurator, output_path: Path, 

21 base_dir: Path, tmp_path: Path, 

22 multi_objective_support: bool = False) -> None: 

23 """Initialize Configurator. 

24 

25 Args: 

26 output_path: Output directory of the Configurator. 

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

28 optimize. 

29 base_dir: Where to execute the configuration 

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

31 multi_objective_support: Whether the configurator supports 

32 multi objective optimization for solvers. 

33 """ 

34 self.output_path = output_path 

35 self.base_dir = base_dir 

36 self.tmp_path = tmp_path 

37 self.multiobjective = multi_objective_support 

38 self.scenario = None 

39 

40 def name(self: Configurator) -> str: 

41 """Return the name of the configurator.""" 

42 return self.__class__.__name__ 

43 

44 @staticmethod 

45 def scenario_class() -> ConfigurationScenario: 

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

47 return ConfigurationScenario 

48 

49 def configure(self: Configurator, 

50 configuration_commands: list[str], 

51 data_target: PerformanceDataFrame, 

52 output: Path, 

53 scenario: ConfigurationScenario, 

54 validation_ids: list[int] = None, 

55 sbatch_options: list[str] = None, 

56 slurm_prepend: str | list[str] | Path = None, 

57 num_parallel_jobs: int = None, 

58 base_dir: Path = None, 

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

60 """Start configuration job. 

61 

62 This method is shared by the configurators and should be called by the 

63 implementation/subclass of the configurator. 

64 

65 Args: 

66 configuration_commands: List of configurator commands to execute 

67 data_target: Performance data to store the results. 

68 output: Output directory. 

69 scenario: ConfigurationScenario to execute. 

70 sbatch_options: List of slurm batch options to use 

71 slurm_prepend: Slurm script to prepend to the sbatch 

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

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

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

75 

76 Returns: 

77 A RunRunner Run object. 

78 """ 

79 runs = [rrr.add_to_queue( 

80 runner=run_on, 

81 cmd=configuration_commands, 

82 name=f"{self.name}: {scenario.solver.name} on {scenario.instance_set.name}", 

83 base_dir=base_dir, 

84 output_path=output, 

85 parallel_jobs=num_parallel_jobs, 

86 sbatch_options=sbatch_options, 

87 prepend=slurm_prepend)] 

88 

89 if validation_ids: 

90 validate = scenario.solver.run_performance_dataframe( 

91 scenario.instance_set, 

92 run_ids=validation_ids, 

93 performance_dataframe=data_target, 

94 cutoff_time=scenario.cutoff_time, 

95 sbatch_options=sbatch_options, 

96 slurm_prepend=slurm_prepend, 

97 log_dir=scenario.validation, 

98 base_dir=base_dir, 

99 dependencies=runs, 

100 job_name=f"{self.name}: Validating {len(validation_ids)} " 

101 f"{scenario.solver.name} Configurations on " 

102 f"{scenario.instance_set.name}", 

103 run_on=run_on, 

104 ) 

105 runs.append(validate) 

106 

107 if run_on == Runner.LOCAL: 

108 print(f"[{self.name}] Running {len(runs)} jobs locally...") 

109 for run in runs: 

110 run.wait() 

111 print(f"[{self.name}] Finished running {len(runs)} jobs locally.") 

112 return runs 

113 

114 @staticmethod 

115 def organise_output(output_source: Path, 

116 output_target: Path, 

117 scenario: ConfigurationScenario, 

118 run_id: int) -> None | str: 

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

120 

121 Args: 

122 output_source: Path to the output file of the configurator run. 

123 output_target: Path to the Performance DataFrame to store result. 

124 scenario: ConfigurationScenario of the configuration. 

125 run_id: ID of the run of the configuration. 

126 """ 

127 raise NotImplementedError 

128 

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

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

131 raise NotImplementedError 

132 

133 

134class ConfigurationScenario: 

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

136 def __init__(self: ConfigurationScenario, 

137 solver: Solver, 

138 instance_set: InstanceSet, 

139 sparkle_objectives: list[SparkleObjective], 

140 parent_directory: Path) -> None: 

141 """Initialize scenario paths and names. 

142 

143 Args: 

144 solver: Solver that should be configured. 

145 instance_set: Instances object for the scenario. 

146 sparkle_objectives: Sparkle Objectives to optimize. 

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

148 """ 

149 self.solver = solver 

150 self.instance_set = instance_set 

151 self.sparkle_objectives = sparkle_objectives 

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

153 

154 if self.instance_set.size == 0: 

155 raise Exception("Cannot configure on an empty instance set " 

156 f"('{instance_set.name}').") 

157 

158 self.directory = parent_directory / self.name 

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

160 self.validation: Path = self.directory / "validation" 

161 self.tmp: Path = self.directory / "tmp" 

162 self.results_directory: Path = self.directory / "results" 

163 

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

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

166 

167 This prepares all the necessary subdirectories related to configuration. 

168 

169 Args: 

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

171 """ 

172 raise NotImplementedError 

173 

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

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

176 

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

178 algo = {configurator_target} {solver_directory} {sparkle_objective} 

179 """ 

180 raise NotImplementedError 

181 

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

183 """Serialize the configuration scenario.""" 

184 raise NotImplementedError 

185 

186 @classmethod 

187 def find_scenario(cls: ConfigurationScenario, 

188 directory: Path, 

189 solver: Solver, 

190 instance_set: InstanceSet) -> ConfigurationScenario: 

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

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

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

194 if not path.exists(): 

195 return None 

196 return cls.from_file(path) 

197 

198 @staticmethod 

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

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

201 raise NotImplementedError