Coverage for sparkle/CLI/help/reporting_scenario.py: 96%

1"""Helper module to manage Sparkle scenarios.""" 

2# Keep in CLI help 

3 

4from __future__ import annotations 

5import configparser 

6from enum import Enum 

7from pathlib import Path 

8from pathlib import PurePath 

9 

10from sparkle.configurator.configurator import ConfigurationScenario 

11from sparkle.solver import Solver 

12from sparkle.instance import Instance_Set, InstanceSet 

13 

14 

15class Scenario(str, Enum): 

16 """Enum of possible execution scenarios for Sparkle.""" 

17 

18 NONE = "NONE" 

19 SELECTION = "SELECTION" 

20 CONFIGURATION = "CONFIGURATION" 

21 PARALLEL_PORTFOLIO = "PARALLEL_PORTFOLIO" 

22 

23 

24class ReportingScenario: 

25 """Class to manage scenarios executed with Sparkle.""" 

26 

27 # ReportingScenario path names and defaults 

28 __reporting_scenario_file = Path("latest_scenario.ini") 

29 __reporting_scenario_dir = Path("Output") 

30 DEFAULT_reporting_scenario_path = Path( 

31 PurePath(__reporting_scenario_dir / __reporting_scenario_file)) 

32 

33 # Constant default values 

34 DEFAULT_latest_scenario = Scenario.NONE 

35 

36 DEFAULT_selection_portfolio_path = Path("") 

37 DEFAULT_selection_test_case_directory = Path("") 

38 

39 DEFAULT_parallel_portfolio_path = Path("") 

40 DEFAULT_parallel_portfolio_instance_list = [] 

41 

42 DEFAULT_config_solver = Path("") 

43 DEFAULT_config_instance_set_train = Path("") 

44 DEFAULT_config_instance_set_test = Path("") 

45 

46 def __init__(self: ReportingScenario) -> None: 

47 """Initialise a ReportingScenario object.""" 

48 # ReportingScenario 'dictionary' in configparser format 

49 self.__scenario = configparser.ConfigParser() 

50 

51 # Initialise scenario in default file path 

52 self.read_scenario_ini() 

53 

54 def read_scenario_ini( 

55 self: ReportingScenario, file_path: Path = DEFAULT_reporting_scenario_path)\ 

56 -> None: 

57 """Read the scenario from an INI file. 

58 

59 Args: 

60 file_path: Path of the INI file for the scenario. Defaults to 

61 DEFAULT_reporting_scenario_path. 

62 """ 

63 # If the file does not exist or is empty set default values 

64 if not Path(file_path).is_file() or Path(file_path).stat().st_size == 0: 

65 self.set_latest_scenario() 

66 self.set_selection_scenario_path() 

67 self.set_selection_test_case_directory() 

68 self.set_parallel_portfolio_path() 

69 self.set_parallel_portfolio_instance_path() 

70 self.set_config_solver() 

71 self.set_config_instance_set_train() 

72 self.set_config_instance_set_test() 

73 

74 # Read file 

75 file_scenario = configparser.ConfigParser() 

76 file_scenario.read(file_path) 

77 

78 # Set internal scenario based on data read from FILE if they were read 

79 # successfully 

80 if file_scenario.sections() != []: 

81 section = "latest" 

82 option_names = ("scenario",) # Comma to make it a tuple 

83 for option in option_names: 

84 if file_scenario.has_option(section, option): 

85 value = Scenario(file_scenario.get(section, option)) 

86 self.set_latest_scenario(value) 

87 file_scenario.remove_option(section, option) 

88 

89 section = "selection" 

90 option_names = ("scenario_path",) # Comma to make it a tuple 

91 for option in option_names: 

92 if file_scenario.has_option(section, option): 

93 value = Path(file_scenario.get(section, option)) 

94 self.set_selection_scenario_path(value) 

95 file_scenario.remove_option(section, option) 

96 

97 option_names = ("test_case_directory",) # Comma to make it a tuple 

98 for option in option_names: 

99 if file_scenario.has_option(section, option): 

100 value = Path(file_scenario.get(section, option)) 

101 self.set_selection_test_case_directory(value) 

102 file_scenario.remove_option(section, option) 

103 

104 section = "configuration" 

105 option_names = ("solver",) # Comma to make it a tuple 

106 for option in option_names: 

107 if file_scenario.has_option(section, option): 

108 value = Path(file_scenario.get(section, option)) 

109 self.set_config_solver(value) 

110 file_scenario.remove_option(section, option) 

111 

112 option_names = ("instance_set_train",) # Comma to make it a tuple 

113 for option in option_names: 

114 if file_scenario.has_option(section, option): 

115 value = Path(file_scenario.get(section, option)) 

116 self.set_config_instance_set_train(value) 

117 file_scenario.remove_option(section, option) 

118 

119 option_names = ("instance_set_test",) # Comma to make it a tuple 

120 for option in option_names: 

121 if file_scenario.has_option(section, option): 

122 value = Path(file_scenario.get(section, option)) 

123 self.set_config_instance_set_test(value) 

124 file_scenario.remove_option(section, option) 

125 

126 option_names = ("scenario_file_path", ) 

127 for option in option_names: 

128 if file_scenario.has_option(section, option): 

129 value = Path(file_scenario.get(section, option)) 

130 self.set_configuration_scenario(value) 

131 file_scenario.remove_option(section, option) 

132 

133 section = "parallel_portfolio" 

134 option_names = ("portfolio_path",) # Comma to make it a tuple 

135 for option in option_names: 

136 if file_scenario.has_option(section, option): 

137 value = Path(file_scenario.get(section, option)) 

138 self.set_parallel_portfolio_path(value) 

139 file_scenario.remove_option(section, option) 

140 

141 option_names = ("instance_path",) # Comma to make it a tuple 

142 for option in option_names: 

143 if file_scenario.has_option(section, option): 

144 value = file_scenario.get(section, option) 

145 self.set_parallel_portfolio_instance_path(value) 

146 file_scenario.remove_option(section, option) 

147 

148 # Report on any unknown settings that were read 

149 sections = file_scenario.sections() 

150 

151 for section in sections: 

152 for option in file_scenario[section]: 

153 print(f'Unrecognised section - option combination:"{section} ' 

154 f'{option}" in file {file_path} ignored') 

155 

156 # Print error if unable to read the scenario file 

157 elif not file_path.exists(): 

158 print(f"WARNING: Failed to read latest scenario from {file_path}. " 

159 "Default values will be used.") 

160 

161 def write_scenario_ini( 

162 self: ReportingScenario, file_path: Path = DEFAULT_reporting_scenario_path)\ 

163 -> None: 

164 """Write the scenario file in INI format. 

165 

166 Args: 

167 file_path: Path of the INI file for the scenario. Defaults to 

168 DEFAULT_reporting_scenario_path. 

169 """ 

170 # Create needed directories if they don't exist 

171 file_dir = file_path.parents[0] 

172 file_dir.mkdir(parents=True, exist_ok=True) 

173 

174 # Write the scenario to file 

175 with Path(str(file_path)).open("w") as scenario_file: 

176 self.__scenario.write(scenario_file) 

177 

178 def __init_section(self: ReportingScenario, section: str) -> None: 

179 """Initialise a section in the scenario file. 

180 

181 Args: 

182 section: Name of the section. 

183 """ 

184 if section not in self.__scenario: 

185 self.__scenario[section] = {} 

186 

187 # Generic setters ### 

188 

189 def path_setter(self: ReportingScenario, section: str, name: str, value: Path)\ 

190 -> None: 

191 """Set a generic Path for the scenario. 

192 

193 Args: 

194 section: Name of the section. 

195 name: Name of the path element. 

196 value: Value of the path given. 

197 """ 

198 if value is not None: 

199 self.__init_section(section) 

200 self.__scenario[section][name] = str(value) 

201 

202 # Generic getters ### 

203 

204 def none_if_empty_path(self: ReportingScenario, path: Path) -> Path: 

205 """Return None if a path is empty or the Path otherwise. 

206 

207 Args: 

208 path: Path value given. 

209 

210 Returns: 

211 None if the given path is empty, the given Path value otherwise. 

212 """ 

213 if str(path) == "" or str(path) == ".": 

214 return None 

215 return path 

216 

217 # Latest settings ### 

218 

219 def set_latest_scenario(self: ReportingScenario, 

220 value: Scenario = DEFAULT_latest_scenario) -> None: 

221 """Set the latest Scenario that was executed.""" 

222 section = "latest" 

223 name = "scenario" 

224 

225 if value is not None: 

226 self.__init_section(section) 

227 self.__scenario[section][name] = value.name 

228 

229 def get_latest_scenario(self: ReportingScenario) -> Scenario: 

230 """Return the latest Scenario that was executed.""" 

231 return Scenario(self.__scenario["latest"]["scenario"]) 

232 

233 # Selection settings ### 

234 

235 def set_selection_scenario_path( 

236 self: ReportingScenario, value: Path = DEFAULT_selection_portfolio_path)\ 

237 -> None: 

238 """Set the path to portfolio selector used for algorithm selection.""" 

239 section = "selection" 

240 name = "scenario_path" 

241 self.path_setter(section, name, value) 

242 

243 def get_selection_scenario_path(self: ReportingScenario) -> Path: 

244 """Return the path to portfolio selector used for algorithm selection.""" 

245 return Path(self.__scenario["selection"]["scenario_path"]) 

246 

247 def set_selection_test_case_directory( 

248 self: ReportingScenario, 

249 value: Path = DEFAULT_selection_test_case_directory) -> None: 

250 """Set the path to the testing set that was used for algorithm selection.""" 

251 section = "selection" 

252 name = "test_case_directory" 

253 self.path_setter(section, name, value) 

254 

255 def get_selection_test_case_directory(self: ReportingScenario) -> str: 

256 """Return the path to the testing set that was used for algorithm selection.""" 

257 try: 

258 path = self.__scenario["selection"]["test_case_directory"] 

259 if Path(path) == Path("."): 

260 path = None 

261 except KeyError: 

262 path = None 

263 return path 

264 

265 # Parallel portfolio settings ### 

266 

267 def set_parallel_portfolio_path( 

268 self: ReportingScenario, 

269 value: Path = DEFAULT_parallel_portfolio_path) -> None: 

270 """Set the path to the parallel portfolio.""" 

271 section = "parallel_portfolio" 

272 name = "portfolio_path" 

273 self.path_setter(section, name, value) 

274 

275 def get_parallel_portfolio_path(self: ReportingScenario) -> Path: 

276 """Return the path to the parallel portfolio.""" 

277 return Path(self.__scenario["parallel_portfolio"]["portfolio_path"]) 

278 

279 def set_parallel_portfolio_instance_path( 

280 self: ReportingScenario, 

281 value: Path = None) -> None: 

282 """Set the instance path used with the parallel portfolio.""" 

283 section = "parallel_portfolio" 

284 name = "instance_path" 

285 self.path_setter(section, name, value) 

286 

287 def get_parallel_portfolio_instance_set(self: ReportingScenario) -> InstanceSet: 

288 """Return the instance list used with the parallel portfolio. 

289 

290 If instance list is empty return None. 

291 """ 

292 if self.__scenario["parallel_portfolio"]["instance_path"] is None: 

293 return None 

294 return Instance_Set(Path(self.__scenario["parallel_portfolio"]["instance_path"])) 

295 

296 # Configuration settings ### 

297 

298 def set_config_solver(self: ReportingScenario, 

299 value: Solver | Path = DEFAULT_config_solver) -> None: 

300 """Set the path to the solver that was configured.""" 

301 section = "configuration" 

302 name = "solver" 

303 if isinstance(value, Solver): 

304 value = value.directory 

305 self.path_setter(section, name, value) 

306 

307 def get_config_solver(self: ReportingScenario) -> Solver: 

308 """Return the path to the solver that was configured.""" 

309 path = self.none_if_empty_path(Path(self.__scenario["configuration"]["solver"])) 

310 if path is not None: 

311 return Solver(path) 

312 return None 

313 

314 def set_config_instance_set_train( 

315 self: ReportingScenario, value: Path = DEFAULT_config_instance_set_train)\ 

316 -> None: 

317 """Set the path to the training instance set used for configuration.""" 

318 section = "configuration" 

319 name = "instance_set_train" 

320 self.path_setter(section, name, value) 

321 

322 def get_config_instance_set_train(self: ReportingScenario) -> InstanceSet: 

323 """Return the path to the training instance set used for configuration.""" 

324 path = self.none_if_empty_path( 

325 Path(self.__scenario["configuration"]["instance_set_train"])) 

326 if path is None: 

327 return None 

328 return Instance_Set(path) 

329 

330 def set_configuration_scenario(self: ReportingScenario, 

331 value: Scenario | Path) -> None: 

332 """Set the path to the scenario that was used for configuration.""" 

333 section = "configuration" 

334 name = "scenario_file_path" 

335 if isinstance(value, ConfigurationScenario): 

336 value = value.scenario_file_path 

337 self.path_setter(section, name, value) 

338 

339 def get_configuration_scenario(self: ReportingScenario, 

340 scenario_class: ConfigurationScenario) -> Scenario: 

341 """Return the path to the scenario that was used for configuration.""" 

342 path = self.none_if_empty_path( 

343 Path(self.__scenario["configuration"]["scenario_file_path"])) 

344 if path is None: 

345 return None 

346 return scenario_class.from_file(path) 

347 

348 def set_config_instance_set_test( 

349 self: ReportingScenario, value: Path = DEFAULT_config_instance_set_test)\ 

350 -> None: 

351 """Set the path to the testing instance set used for configuration.""" 

352 section = "configuration" 

353 name = "instance_set_test" 

354 self.path_setter(section, name, value) 

355 

356 def get_config_instance_set_test(self: ReportingScenario) -> InstanceSet: 

357 """Return the path to the testing instance set used for configuration.""" 

358 path = self.none_if_empty_path( 

359 Path(self.__scenario["configuration"]["instance_set_test"])) 

360 if path is None: 

361 return None 

362 return Instance_Set(path)