Coverage for src/sparkle/CLI/add_solver.py: 80%

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 

20from sparkle.CLI.help import jobs as jobs_help 

21 

22 

23def parser_function() -> argparse.ArgumentParser: 

24 """Define the command line arguments.""" 

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

26 parser.add_argument( 

27 *ac.DeterministicArgument.names, **ac.DeterministicArgument.kwargs 

28 ) 

29 parser.add_argument( 

30 *ac.SolutionVerifierArgument.names, **ac.SolutionVerifierArgument.kwargs 

31 ) 

32 parser.add_argument( 

33 *ac.NicknameSolverArgument.names, **ac.NicknameSolverArgument.kwargs 

34 ) 

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

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

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

38 return parser 

39 

40 

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

42 """Main function of the command.""" 

43 # Log command call 

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

45 check_for_initialise() 

46 

47 # Define command line arguments 

48 parser = parser_function() 

49 

50 # Process command line arguments 

51 args = parser.parse_args(argv) 

52 solver_source = Path(args.solver_path) 

53 deterministic = args.deterministic 

54 solution_verifier = args.solution_verifier 

55 

56 if not solver_source.exists(): 

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

58 sys.exit(-1) 

59 

60 # Make sure it is pointing to the verifiers module 

61 if solution_verifier: 

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

63 solution_verifier = ( 

64 verifiers.SolutionFileVerifier.__name__, 

65 solution_verifier, 

66 ) 

67 elif solution_verifier not in verifiers.mapping: 

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

69 sys.exit(-1) 

70 

71 nickname = args.nickname 

72 

73 if args.run_checks: 

74 print("Running checks...") 

75 solver = Solver(Path(solver_source)) 

76 if solver.pcs_file is None: 

77 print( 

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

79 "is not valid for configuration." 

80 ) 

81 else: 

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

83 if solver.read_pcs_file(): 

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

85 else: 

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

87 

88 wrapper_path = solver.directory / solver.wrapper 

89 if not wrapper_path.is_file(): 

90 print( 

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

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

93 ) 

94 sys.exit(-1) 

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

96 print( 

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

98 f" does not have execution rights set!" 

99 ) 

100 sys.exit(-1) 

101 

102 # Start add solver 

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

104 if solver_directory.exists(): 

105 print( 

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

107 ) 

108 sys.exit(-1) 

109 if args.no_copy: 

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

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

112 raise PermissionError( 

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

114 ) 

115 solver_directory.symlink_to(solver_source.absolute()) 

116 else: 

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

118 solver_directory.mkdir(parents=True) 

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

120 

121 # Save the deterministic bool in the solver 

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

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

124 

125 # Add RunSolver executable to the solver 

126 runsolver_path = gv.settings().DEFAULT_runsolver_exec 

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

128 print( 

129 "Warning! RunSolver executable detected in Solver " 

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

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

132 ) 

133 

134 if runsolver_path.exists(): 

135 runsolver_target = solver_directory / runsolver_path.name 

136 shutil.copyfile(runsolver_path, runsolver_target) 

137 runsolver_target.chmod(runsolver_target.stat().st_mode | stat.S_IEXEC) 

138 else: 

139 print("Warning! RunSolver does not exists. Falling back to PyRunSolver.") 

140 

141 jobs_help.check_running_waiting_jobs( 

142 gv.settings().DEFAULT_log_output, 

143 ) 

144 

145 performance_data = PerformanceDataFrame( 

146 gv.settings().DEFAULT_performance_data_path, 

147 objectives=gv.settings().objectives, 

148 ) 

149 performance_data.add_solver(str(solver_directory)) 

150 performance_data.save_csv() 

151 

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

153 

154 if nickname is not None: 

155 sfh.add_remove_platform_item( 

156 solver_directory, 

157 gv.solver_nickname_list_path, 

158 gv.file_storage_data_mapping[gv.solver_nickname_list_path], 

159 key=nickname, 

160 ) 

161 

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

163 if solver.pcs_file is not None: 

164 # Generate missing PCS files 

165 print("Checking for missing PCS files to generate...") 

166 if solver.get_pcs_file_type(PCSConvention.IRACE) is None: 

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

168 print("\t- Generating IRACE done!") 

169 if solver.get_pcs_file_type(PCSConvention.ParamILS) is None: 

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

171 print("\t- Generating ParamILS done!") 

172 

173 print(f"Solver {solver.name} added to platform!") 

174 

175 # Write used settings to file 

176 gv.settings().write_used_settings() 

177 sys.exit(0) 

178 

179 

180if __name__ == "__main__": 

181 main(sys.argv[1:])