Coverage for src / sparkle / CLI / wrap.py: 28%

180 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-01-21 15:31 +0000

1"""Command to wrap users' Solvers / Feature extractors for Sparkle.""" 

2 

3import sys 

4import subprocess 

5import re 

6import argparse 

7from pathlib import Path 

8 

9 

10from sparkle.solver import Solver 

11from sparkle.selector.extractor import Extractor 

12from sparkle.platform import Settings 

13 

14from sparkle.CLI.help import logging as sl 

15from sparkle.CLI.help import argparse_custom as ac 

16from sparkle.CLI.help import global_variables as gv 

17 

18import ConfigSpace 

19import numpy as np 

20from ConfigSpace.types import NotSet, i64 

21 

22 

23def cli_to_configspace( 

24 input_data: str, name: str = None 

25) -> ConfigSpace.ConfigurationSpace: 

26 """Attempts to process CLI help string to a ConfigSpace representation. 

27 

28 Args: 

29 input_data: CLI help string containing parameter data. 

30 name: Name to give to the ConfigSpace 

31 

32 Returns: 

33 ConfigSpace object. 

34 """ 

35 space = ConfigSpace.ConfigurationSpace(name=name) 

36 

37 parameter_data = [] 

38 first_group = False 

39 for line in input_data.split("\n"): 

40 line = line.strip() 

41 if line == "": 

42 first_group = False 

43 if line.startswith("-"): 

44 first_group = True 

45 parameter_data.append(line) 

46 elif not first_group: 

47 continue 

48 else: 

49 parameter_data[-1] = parameter_data[-1] + " " + line 

50 if not parameter_data: 

51 return space 

52 name_pattern = r"(?<!\S)--?[\w-]+" 

53 int_min, int_max = ( 

54 int(np.iinfo(i64).min / 10), 

55 int(np.iinfo(i64).max / 10), 

56 ) # ConfigSpace bug on positive max size? Also causes an error during sampling 

57 float_min, float_max = -sys.maxsize, sys.maxsize 

58 print( 

59 "For each parameter we need to know the parameter type, please choose for each found out of the following:" 

60 ) 

61 print( 

62 "\t- Integer\n\t- Float\n\t- Ordinal\n\t- Categorical\n\t- Boolean\n\t- Empty/Skip [Do not add this parameter]\n" 

63 ) 

64 for parameter in parameter_data: 

65 matches = re.findall(name_pattern, parameter) 

66 for match in matches: 

67 name = match.strip("-") 

68 if ( 

69 len(name) == 1 and len(matches) > 1 

70 ): # Short version of the parameter, continue 

71 continue 

72 break 

73 

74 print(f"\nParameter [{name}]: ") 

75 print(f"Description: {parameter}") 

76 value = input("Please specify the parameter type: ") 

77 if value.lower() in ["", "empty", "skip"]: 

78 print("> Skipping parameter...") 

79 continue 

80 default = input( 

81 "Please specify the parameter default value (Empty for not set): " 

82 ) 

83 if default == "": 

84 default = NotSet 

85 

86 match value.lower(): 

87 case "integer" | "int" | "1": # Integer 

88 lower, upper = None, None 

89 while lower is None or upper is None: 

90 user_input = input( 

91 "Please specify the integer lower and upper limit separated by a comma (,). Empty defaults to -max / max: " 

92 ) 

93 if "," in user_input: 

94 lower, upper = user_input.split(",", maxsplit=1) 

95 lower, upper = lower.strip(), upper.strip() 

96 lower = int_min if lower == "" else i64(lower) 

97 upper = int_max if upper == "" else i64(upper) 

98 default = int(default) if default != NotSet else None 

99 log = ( 

100 input("Should the values be sampled on a log-scale? (y/n): ").lower() 

101 == "y" 

102 ) 

103 try: 

104 space.add( 

105 ConfigSpace.UniformIntegerHyperparameter( 

106 name=name, 

107 lower=lower, 

108 upper=upper, 

109 default_value=default, 

110 log=log, 

111 meta=parameter, 

112 ) 

113 ) 

114 except Exception as e: 

115 print("The following exception occured: ", e) 

116 print("Continuing to the next parameter...") 

117 continue 

118 case "float" | "2": # Float 

119 lower, upper = None, None 

120 while lower is None or upper is None: 

121 user_input = input( 

122 "Please specify the float lower and upper limit separated by a comma (,). Empty defaults to -max / max: " 

123 ) 

124 if "," in user_input: 

125 lower, upper = user_input.split(",", maxsplit=1) 

126 lower, upper = lower.strip(), upper.strip() 

127 lower = float_min if lower == "" else float(lower) 

128 upper = float_max if upper == "" else float(upper) 

129 default = float(default) if default != NotSet else None 

130 log = ( 

131 input("Should the values be sampled on a log-scale? (y/n): ").lower() 

132 == "y" 

133 ) 

134 try: 

135 space.add( 

136 ConfigSpace.UniformFloatHyperparameter( 

137 name=name, 

138 lower=lower, 

139 upper=upper, 

140 default_value=default, 

141 log=log, 

142 meta=parameter, 

143 ) 

144 ) 

145 except Exception as e: 

146 print("The following exception occured: ", e) 

147 print("Continuing to the next parameter...") 

148 continue 

149 case "ordinal" | "ord" | "3": # Ordinal 

150 print( 

151 "Please specify the ordinal ascending sequence separated by a comma (,): ", 

152 end="", 

153 ) 

154 sequence = [ 

155 s.strip() 

156 for s in input( 

157 "Please specify the ordinal ascending sequence separated by a comma (,): " 

158 ).split(",") 

159 ] 

160 if default != NotSet: 

161 while default not in sequence: 

162 default = input( 

163 "Please specify the default value from the sequence: " 

164 ) 

165 elif default not in sequence: 

166 print( 

167 "The default value is not in the sequence, please specify a new default value, or leave empty to add it to the sequence." 

168 ) 

169 while default != "" or default not in sequence: 

170 default = input("Default value: ") 

171 if default == "": 

172 sequence.append(default) 

173 try: 

174 space.add( 

175 ConfigSpace.OrdinalHyperparameter( 

176 name=name, 

177 sequence=sequence, 

178 default_value=default, 

179 meta=parameter, 

180 ) 

181 ) 

182 except Exception as e: 

183 print("The following exception occured: ", e) 

184 print("Continuing to the next parameter...") 

185 continue 

186 case "categorical" | "cat" | "4": # Categorical 

187 choices = [ 

188 s.strip() 

189 for s in input( 

190 "Please specify the categorical options separated by a comma (,): " 

191 ).split(",") 

192 ] 

193 choices = [option.strip() for option in choices] 

194 if default != NotSet: 

195 while default not in choices: 

196 default = input( 

197 "Please specify the default value from the choices: " 

198 ) 

199 elif default not in choices: 

200 print( 

201 "The default value is not in the choices, please specify a new default value, or leave empty to add it to the choices." 

202 ) 

203 while default != "" or default not in choices: 

204 default = input("Default value: ") 

205 if default == "": 

206 choices.append(default) 

207 try: 

208 space.add( 

209 ConfigSpace.CategoricalHyperparameter( 

210 name=name, 

211 choices=choices, 

212 default_value=default, 

213 meta=parameter, 

214 ) 

215 ) 

216 except Exception as e: 

217 print("The following exception occured: ", e) 

218 print("Continuing to the next parameter...") 

219 continue 

220 case "boolean" | "bool" | "5": # Boolean 

221 if default != NotSet: 

222 while default not in ["True", "False"]: 

223 default = input( 

224 "Please specify the default value as True/False: " 

225 ) 

226 space.add( 

227 ConfigSpace.CategoricalHyperparameter( 

228 name=name, 

229 choices=["True", "False"], 

230 default_value=default, 

231 ) 

232 ) 

233 case _: # Skip 

234 continue 

235 return space 

236 

237 

238def parser_function() -> argparse.ArgumentParser: 

239 """Define the command line arguments.""" 

240 parser = argparse.ArgumentParser( 

241 description="Command to wrap input solvers and feature extractors." 

242 "Specify a path to the directory containing your solvers." 

243 ) 

244 parser.add_argument(*ac.WrapTypeArgument.names, **ac.WrapTypeArgument.kwargs) 

245 parser.add_argument(*ac.WrapPathArgument.names, **ac.WrapPathArgument.kwargs) 

246 parser.add_argument(*ac.WrapTargetArgument.names, **ac.WrapTargetArgument.kwargs) 

247 parser.add_argument( 

248 *ac.WrapGeneratePCSArgument.names, **ac.WrapGeneratePCSArgument.kwargs 

249 ) 

250 return parser 

251 

252 

253def main(argv: list[str]) -> None: 

254 """Main entry point of the Check command.""" 

255 parser = parser_function() 

256 args = parser.parse_args(argv) 

257 settings = gv.settings(args) 

258 

259 # Log command call 

260 sl.log_command(sys.argv, settings.random_state) 

261 type_map = { 

262 "extractor": Extractor, 

263 "feature-extractor": Extractor, 

264 "solver": Solver, 

265 "Extractor": Extractor, 

266 "Feature-Extractor": Extractor, 

267 "Solver": Solver, 

268 "FeatureExtractor": Extractor, 

269 } 

270 

271 if args.type not in type_map: 

272 options_text = "\n".join([f"\t- {value}" for value in type_map.keys()]) 

273 raise ValueError( 

274 f"Unknown type {args.type}. Please choose from:\n{options_text}" 

275 ) 

276 type = type_map[args.type] 

277 

278 print(f"Wrapping {type.__name__} in directory {args.path} ...") 

279 path: Path = args.path 

280 if not path.exists(): 

281 raise ValueError(f"Directory {path} does not exist.") 

282 

283 target_path: Path = args.target 

284 if ( 

285 path not in target_path.parents 

286 ): # Allow the user to flexibly specify the target (as path/executable or just executable) 

287 target_path = path / args.target 

288 

289 if not target_path.exists(): 

290 raise ValueError(f"Target executable {target_path} does not exist.") 

291 

292 if type == Solver: 

293 target_wrapper = path / ( 

294 Solver._wrapper_file + Settings.DEFAULT_solver_wrapper_template.suffix 

295 ) 

296 if target_wrapper.exists(): 

297 print(f"WARNING: Wrapper {target_wrapper} already exists. Skipping...") 

298 else: 

299 template_data = Settings.DEFAULT_solver_wrapper_template.open().read() 

300 template_data = template_data.replace( 

301 "@@@YOUR_EXECUTABLE_HERE@@@", str(target_path.relative_to(path)) 

302 ) 

303 target_wrapper.open("w").write(template_data) 

304 target_wrapper.chmod(0o755) # Set read and execution rights for all 

305 if args.generate_pcs: 

306 pcs_file: Path = path / "sparkle_PCS.yaml" 

307 if pcs_file.exists(): 

308 print(f"WARNING: PCS file {pcs_file} already exists. Skipping...") 

309 else: 

310 solver_call = subprocess.run( 

311 [str(target_path), "--help"], capture_output=True 

312 ) 

313 input_data = ( 

314 solver_call.stdout.decode("utf-8") 

315 + "\n" 

316 + solver_call.stderr.decode("utf-8") 

317 ) 

318 space = cli_to_configspace(input_data, name=target_path.stem) 

319 if len(space) == 0: # Failed to extract anything 

320 print( 

321 "Could not extract any parameters from the executable. No PCS file was created." 

322 ) 

323 sys.exit(-1) 

324 space.to_yaml(pcs_file) 

325 elif type == Extractor: 

326 raise NotImplementedError("Feature extractor wrapping not implemented yet.") 

327 

328 sys.exit(0) 

329 

330 

331if __name__ == "__main__": 

332 main(sys.argv[1:])