Coverage for sparkle/CLI/add_solver.py: 81%

94 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2025-09-29 10:17 +0000

1#!/usr/bin/env python3 

2"""Sparkle command to add a solver to the Sparkle platform.""" 

3 

4import os 

5import stat 

6import sys 

7import argparse 

8import shutil 

9from pathlib import Path 

10 

11from sparkle.tools.parameters import PCSConvention 

12 

13from sparkle.platform import file_help as sfh 

14from sparkle.CLI.help import global_variables as gv 

15from sparkle.structures import PerformanceDataFrame 

16from sparkle.solver import Solver, verifiers 

17from sparkle.CLI.help import logging as sl 

18from sparkle.CLI.initialise import check_for_initialise 

19from sparkle.CLI.help import argparse_custom as ac 

20 

21 

22def parser_function() -> argparse.ArgumentParser: 

23 """Define the command line arguments.""" 

24 parser = argparse.ArgumentParser(description="Add a solver to the Sparkle platform.") 

25 parser.add_argument( 

26 *ac.DeterministicArgument.names, **ac.DeterministicArgument.kwargs 

27 ) 

28 parser.add_argument( 

29 *ac.SolutionVerifierArgument.names, **ac.SolutionVerifierArgument.kwargs 

30 ) 

31 parser.add_argument( 

32 *ac.NicknameSolverArgument.names, **ac.NicknameSolverArgument.kwargs 

33 ) 

34 parser.add_argument(*ac.SolverPathArgument.names, **ac.SolverPathArgument.kwargs) 

35 parser.add_argument(*ac.SkipChecksArgument.names, **ac.SkipChecksArgument.kwargs) 

36 parser.add_argument(*ac.NoCopyArgument.names, **ac.NoCopyArgument.kwargs) 

37 return parser 

38 

39 

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

41 """Main function of the command.""" 

42 # Log command call 

43 sl.log_command(sys.argv, gv.settings().random_state) 

44 check_for_initialise() 

45 

46 # Define command line arguments 

47 parser = parser_function() 

48 

49 # Process command line arguments 

50 args = parser.parse_args(argv) 

51 solver_source = Path(args.solver_path) 

52 deterministic = args.deterministic 

53 solution_verifier = args.solution_verifier 

54 

55 if not solver_source.exists(): 

56 print(f'Solver path "{solver_source}" does not exist!') 

57 sys.exit(-1) 

58 

59 # Make sure it is pointing to the verifiers module 

60 if solution_verifier: 

61 if Path(solution_verifier).is_file(): # File verifier 

62 solution_verifier = ( 

63 verifiers.SolutionFileVerifier.__name__, 

64 solution_verifier, 

65 ) 

66 elif solution_verifier not in verifiers.mapping: 

67 print(f"ERROR: Unknown solution verifier {solution_verifier}!") 

68 sys.exit(-1) 

69 

70 nickname = args.nickname 

71 

72 if args.run_checks: 

73 print("Running checks...") 

74 solver = Solver(Path(solver_source)) 

75 if solver.pcs_file is None: 

76 print( 

77 "None or multiple .pcs files found. Solver " 

78 "is not valid for configuration." 

79 ) 

80 else: 

81 print(f"PCS file detected: {solver.pcs_file.name}. ", end="") 

82 if solver.read_pcs_file(): 

83 print("Can read the pcs file.") 

84 else: 

85 print("WARNING: Can not read the provided pcs file format.") 

86 

87 wrapper_path = solver.directory / solver.wrapper 

88 if not wrapper_path.is_file(): 

89 print( 

90 f"ERROR: Solver {solver_source.name} does not have a solver wrapper " 

91 f"(Missing file {solver.wrapper})." 

92 ) 

93 sys.exit(-1) 

94 elif not os.access(wrapper_path, os.X_OK): 

95 print( 

96 f"ERROR: Solver {solver_source.name} wrapper file {solver.wrapper} " 

97 f" does not have execution rights set!" 

98 ) 

99 sys.exit(-1) 

100 

101 # Start add solver 

102 solver_directory = gv.settings().DEFAULT_solver_dir / solver_source.name 

103 if solver_directory.exists(): 

104 print( 

105 f"ERROR: Solver {solver_source.name} already exists! Can not add new solver." 

106 ) 

107 sys.exit(-1) 

108 if args.no_copy: 

109 print(f"Creating symbolic link from {solver_source} to {solver_directory}...") 

110 if not os.access(solver_source, os.W_OK): 

111 raise PermissionError( 

112 "Sparkle does not have the right to write to the destination folder." 

113 ) 

114 solver_directory.symlink_to(solver_source.absolute()) 

115 else: 

116 print(f"Copying {solver_source.name} to platform...") 

117 solver_directory.mkdir(parents=True) 

118 shutil.copytree(solver_source, solver_directory, dirs_exist_ok=True) 

119 

120 # Save the deterministic bool in the solver 

121 with (solver_directory / Solver.meta_data).open("w+") as fout: 

122 fout.write(str({"deterministic": deterministic, "verifier": solution_verifier})) 

123 

124 # Add RunSolver executable to the solver 

125 runsolver_path = gv.settings().DEFAULT_runsolver_exec 

126 if runsolver_path.name in [file.name for file in solver_directory.iterdir()]: 

127 print( 

128 "Warning! RunSolver executable detected in Solver " 

129 f"{solver_source.name}. This will be replaced with " 

130 f"Sparkle's version of RunSolver. ({runsolver_path})" 

131 ) 

132 runsolver_target = solver_directory / runsolver_path.name 

133 shutil.copyfile(runsolver_path, runsolver_target) 

134 runsolver_target.chmod(os.stat(runsolver_target).st_mode | stat.S_IEXEC) 

135 

136 performance_data = PerformanceDataFrame( 

137 gv.settings().DEFAULT_performance_data_path, 

138 objectives=gv.settings().objectives, 

139 ) 

140 performance_data.add_solver(str(solver_directory)) 

141 performance_data.save_csv() 

142 

143 print(f"Adding solver {solver_source.name} done!") 

144 

145 if nickname is not None: 

146 sfh.add_remove_platform_item( 

147 solver_directory, 

148 gv.solver_nickname_list_path, 

149 gv.file_storage_data_mapping[gv.solver_nickname_list_path], 

150 key=nickname, 

151 ) 

152 

153 solver = Solver(solver_directory) # Recreate solver from its new directory 

154 if solver.pcs_file is not None: 

155 # Generate missing PCS files 

156 # TODO: Only generate missing files 

157 print("Generating missing PCS files...") 

158 solver.port_pcs(PCSConvention.IRACE) # Create PCS file for IRACE 

159 print("Generating IRACE done!") 

160 solver.port_pcs(PCSConvention.ParamILS) # Create PCS file for ParamILS 

161 print("Generating ParamILS done!") 

162 

163 # Write used settings to file 

164 gv.settings().write_used_settings() 

165 sys.exit(0) 

166 

167 

168if __name__ == "__main__": 

169 main(sys.argv[1:])