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 Args: 

63 configuration_commands: List of configurator commands to execute 

64 data_target: Performance data to store the results. 

65 output: Output directory. 

66 scenario: ConfigurationScenario to execute. 

67 sbatch_options: List of slurm batch options to use 

68 slurm_prepend: Slurm script to prepend to the sbatch 

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

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

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

72 

73 Returns: 

74 A RunRunner Run object. 

75 """ 

76 runs = [rrr.add_to_queue( 

77 runner=run_on, 

78 cmd=configuration_commands, 

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

80 base_dir=base_dir, 

81 output_path=output, 

82 parallel_jobs=num_parallel_jobs, 

83 sbatch_options=sbatch_options, 

84 prepend=slurm_prepend)] 

85 

86 if validation_ids: 

87 validate = scenario.solver.run_performance_dataframe( 

88 scenario.instance_set, 

89 run_ids=validation_ids, 

90 performance_dataframe=data_target, 

91 cutoff_time=scenario.cutoff_time, 

92 sbatch_options=sbatch_options, 

93 log_dir=scenario.validation, 

94 base_dir=base_dir, 

95 dependencies=runs, 

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

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

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

99 run_on=run_on, 

100 ) 

101 runs.append(validate) 

102 

103 if run_on == Runner.LOCAL: 

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

105 for run in runs: 

106 run.wait() 

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

108 return runs 

109 

110 @staticmethod 

111 def organise_output(output_source: Path, 

112 output_target: Path, 

113 scenario: ConfigurationScenario, 

114 run_id: int) -> None | str: 

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

116 

117 Args: 

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

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

120 scenario: ConfigurationScenario of the configuration. 

121 run_id: ID of the run of the configuration. 

122 """ 

123 raise NotImplementedError 

124 

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

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

127 raise NotImplementedError 

128 

129 

130class ConfigurationScenario: 

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

132 def __init__(self: ConfigurationScenario, 

133 solver: Solver, 

134 instance_set: InstanceSet, 

135 sparkle_objectives: list[SparkleObjective], 

136 parent_directory: Path) -> None: 

137 """Initialize scenario paths and names. 

138 

139 Args: 

140 solver: Solver that should be configured. 

141 instance_set: Instances object for the scenario. 

142 sparkle_objectives: Sparkle Objectives to optimize. 

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

144 """ 

145 self.solver = solver 

146 self.instance_set = instance_set 

147 self.sparkle_objectives = sparkle_objectives 

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

149 

150 if self.instance_set.size == 0: 

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

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

153 

154 self.directory = parent_directory / self.name 

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

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

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

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

159 

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

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

162 

163 This prepares all the necessary subdirectories related to configuration. 

164 

165 Args: 

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

167 """ 

168 raise NotImplementedError 

169 

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

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

172 

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

174 algo = {configurator_target} {solver_directory} {sparkle_objective} 

175 """ 

176 raise NotImplementedError 

177 

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

179 """Serialize the configuration scenario.""" 

180 raise NotImplementedError 

181 

182 @classmethod 

183 def find_scenario(cls: ConfigurationScenario, 

184 directory: Path, 

185 solver: Solver, 

186 instance_set: InstanceSet) -> ConfigurationScenario: 

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

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

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

190 if not path.exists(): 

191 return None 

192 return cls.from_file(path) 

193 

194 @staticmethod 

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

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

197 raise NotImplementedError