Coverage for sparkle/CLI/run_ablation.py: 76%

1#!/usr/bin/env python3 

2"""Sparkle command to execute ablation analysis.""" 

3 

4import argparse 

5import sys 

6 

7from runrunner.base import Runner 

8 

9from sparkle.configurator import AblationScenario 

10from sparkle.CLI.help import global_variables as gv 

11from sparkle.CLI.help import logging as sl 

12from sparkle.platform.settings_objects import Settings 

13from sparkle.solver import Solver 

14from sparkle.structures import PerformanceDataFrame 

15from sparkle.instance import Instance_Set, InstanceSet 

16from sparkle.CLI.help import argparse_custom as ac 

17from sparkle.CLI.initialise import check_for_initialise 

18from sparkle.CLI.help.nicknames import resolve_object_name 

19 

20 

21def parser_function() -> argparse.ArgumentParser: 

22 """Define the command line arguments.""" 

23 parser = argparse.ArgumentParser( 

24 description="Runs parameter importance between the default and configured " 

25 "parameters with ablation. This command requires a finished " 

26 "configuration for the solver instance pair.", 

27 epilog="Note that if no test instance set is given, the validation is performed" 

28 " on the training set.", 

29 ) 

30 parser.add_argument("--solver", required=False, type=str, help="path to solver") 

31 parser.add_argument( 

32 *ac.InstanceSetTrainOptionalArgument.names, 

33 **ac.InstanceSetTrainOptionalArgument.kwargs, 

34 ) 

35 parser.add_argument( 

36 *ac.InstanceSetTestAblationArgument.names, 

37 **ac.InstanceSetTestAblationArgument.kwargs, 

38 ) 

39 # Settings arguments 

40 parser.add_argument(*ac.SettingsFileArgument.names, **ac.SettingsFileArgument.kwargs) 

41 parser.add_argument( 

42 *Settings.OPTION_ablation_racing.args, **Settings.OPTION_ablation_racing.kwargs 

43 ) 

44 parser.add_argument(*Settings.OPTION_run_on.args, **Settings.OPTION_run_on.kwargs) 

45 parser.set_defaults(ablation_settings_help=False) 

46 return parser 

47 

48 

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

50 """Main function to run ablation analysis.""" 

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

52 check_for_initialise() 

53 

54 if not AblationScenario.check_requirements(verbose=True): 

55 print("Ablation Analysis is not available.") 

56 if not AblationScenario.ablation_executable.exists(): 

57 print("Would you like to download it? (Y/n)") 

58 if input().lower().strip() == "y": 

59 AblationScenario.download_requirements() 

60 else: 

61 sys.exit() 

62 else: 

63 print("Check that Java is available on your system.") 

64 sys.exit(-1) 

65 

66 # Define command line arguments 

67 parser = parser_function() 

68 

69 # Process command line arguments 

70 args = parser.parse_args(argv) 

71 settings = gv.settings(args) 

72 

73 # Compare current settings to latest.ini 

74 prev_settings = Settings(Settings.DEFAULT_previous_settings_path) 

75 Settings.check_settings_changes(settings, prev_settings) 

76 

77 solver = resolve_object_name( 

78 args.solver, gv.solver_nickname_mapping, settings.DEFAULT_solver_dir, Solver 

79 ) 

80 if solver is None: 

81 print(f"Could not resolve Solver path/name {args.solver}!") 

82 print([p for p in settings.DEFAULT_solver_dir.iterdir()]) 

83 sys.exit(-1) 

84 

85 instance_set_train: InstanceSet = resolve_object_name( 

86 args.instance_set_train, 

87 gv.file_storage_data_mapping[gv.instances_nickname_path], 

88 settings.DEFAULT_instance_dir, 

89 Instance_Set, 

90 ) 

91 instance_set_test = resolve_object_name( 

92 args.instance_set_test, 

93 gv.file_storage_data_mapping[gv.instances_nickname_path], 

94 settings.DEFAULT_instance_dir, 

95 Instance_Set, 

96 ) 

97 

98 configurator = settings.configurator 

99 output_path = settings.get_configurator_output_path(configurator) 

100 

101 config_scenario = configurator.scenario_class().find_scenario( 

102 output_path, solver, instance_set_train 

103 ) 

104 

105 performance_data = PerformanceDataFrame(settings.DEFAULT_performance_data_path) 

106 if config_scenario is None: 

107 print( 

108 "No configuration scenario found for combination:\n" 

109 f"{configurator.name} {solver.name} {instance_set_train.name}" 

110 ) 

111 sys.exit(-1) 

112 best_configuration_key, _ = performance_data.best_configuration( 

113 str(solver.directory), 

114 config_scenario.sparkle_objective, 

115 instances=instance_set_train.instance_names, 

116 ) 

117 best_configuration = performance_data.get_full_configuration( 

118 str(solver.directory), best_configuration_key 

119 ) 

120 if instance_set_test is None: 

121 instance_set_test = instance_set_train 

122 

123 if not config_scenario.results_directory.is_dir(): 

124 print( 

125 "Error: No configuration results found for the given solver and training" 

126 " instance set. Ablation needs to have a target configuration. " 

127 "Please finish configuration first." 

128 ) 

129 sys.exit(-1) 

130 else: 

131 print("Configuration exists!") 

132 

133 ablation_scenario = AblationScenario( 

134 config_scenario, 

135 instance_set_test, 

136 cutoff_length=settings.smac2_target_cutoff_length, # NOTE: SMAC2 

137 concurrent_clis=settings.slurm_jobs_in_parallel, 

138 best_configuration=best_configuration, 

139 ablation_racing=settings.ablation_racing_flag, 

140 ) 

141 

142 # Create scenario 

143 ablation_scenario.create_scenario(override_dirs=True) 

144 

145 print("Submiting ablation run...") 

146 run_on = settings.run_on 

147 runs = ablation_scenario.submit_ablation( 

148 log_dir=sl.caller_log_dir, 

149 sbatch_options=settings.sbatch_settings, 

150 run_on=run_on, 

151 ) 

152 

153 if run_on == Runner.LOCAL: 

154 print("Ablation analysis finished!") 

155 else: 

156 job_id_str = ",".join([run.run_id for run in runs]) 

157 print(f"Ablation analysis running through Slurm with job id(s): {job_id_str}") 

158 sys.exit(0) 

159 

160 

161if __name__ == "__main__": 

162 main(sys.argv[1:])