Coverage for src/sparkle/CLI/run_portfolio_selector.py: 30%

1#!/usr/bin/env python3 

2"""Sparkle command to execute a portfolio selector.""" 

3 

4import sys 

5import argparse 

6 

7from runrunner import Runner 

8 

9from sparkle.CLI.help import global_variables as gv 

10from sparkle.CLI.help import logging as sl 

11from sparkle.platform.settings_objects import Settings 

12from sparkle.CLI.help import argparse_custom as ac 

13from sparkle.structures import FeatureDataFrame 

14from sparkle.CLI.initialise import check_for_initialise 

15from sparkle.CLI.help.nicknames import resolve_object_name 

16from sparkle.instance import Instance_Set, InstanceSet 

17from sparkle.CLI.compute_features import compute_features 

18from sparkle.selector import SelectionScenario, Extractor 

19from sparkle.CLI.help import jobs as jobs_help 

20 

21 

22def parser_function() -> argparse.ArgumentParser: 

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

24 parser = argparse.ArgumentParser( 

25 description="Run a portfolio selector on instance (set): Determine which solver " 

26 "is most likely to perform well and run it on the instance (set)." 

27 ) 

28 parser.add_argument( 

29 *ac.SelectionScenarioArgument.names, **ac.SelectionScenarioArgument.kwargs 

30 ) 

31 parser.add_argument( 

32 *ac.InstanceSetRequiredArgument.names, **ac.InstanceSetRequiredArgument.kwargs 

33 ) 

34 # Settings arguments 

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

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

37 return parser 

38 

39 

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

41 """Main function of the run portfolio selector command.""" 

42 # Define command line arguments 

43 parser = parser_function() 

44 # Process command line arguments 

45 args = parser.parse_args(argv) 

46 settings = gv.settings(args) 

47 

48 # Log command call 

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

50 check_for_initialise() 

51 

52 # Compare current settings to latest.ini 

53 prev_settings = Settings(Settings.DEFAULT_previous_settings_path) 

54 Settings.check_settings_changes(settings, prev_settings) 

55 

56 data_set: InstanceSet = resolve_object_name( 

57 args.instance, 

58 gv.file_storage_data_mapping[gv.instances_nickname_path], 

59 settings.DEFAULT_instance_dir, 

60 Instance_Set, 

61 ) 

62 

63 if data_set is None: 

64 print( 

65 "ERROR: The instance (set) could not be found. Please make sure the " 

66 "path is correct." 

67 ) 

68 sys.exit(-1) 

69 

70 run_on = settings.run_on 

71 selector_scenario = SelectionScenario.from_file(args.selection_scenario) 

72 jobs_help.check_running_waiting_jobs( 

73 settings.DEFAULT_log_output, 

74 ) 

75 

76 # Create a new feature dataframe for this run, compute the features 

77 test_case_path = selector_scenario.directory / data_set.name 

78 test_case_path.mkdir(exist_ok=True) 

79 feature_dataframe = FeatureDataFrame(test_case_path / "feature_data.csv") 

80 feature_dataframe.remove_instance(feature_dataframe.instance_pairs) 

81 

82 for extractor_name in selector_scenario.feature_extractors: 

83 extractor = resolve_object_name( 

84 extractor_name, 

85 gv.file_storage_data_mapping[gv.instances_nickname_path], 

86 settings.DEFAULT_extractor_dir, 

87 Extractor, 

88 ) 

89 feature_dataframe.add_extractor(extractor_name, extractor.features) 

90 

91 feature_dataframe.add_instance(data_set.instance_pairs) 

92 feature_dataframe.save_csv() 

93 feature_runs = compute_features( 

94 feature_dataframe, 

95 recompute=False, 

96 run_on=run_on, 

97 instance_sets=[data_set], 

98 ) 

99 

100 # Results need to be stored in the performance data object of the scenario: 

101 # Add the instance set to it 

102 selector_scenario.selector_performance_data.add_instance(data_set.instance_pairs) 

103 selector_scenario.selector_performance_data.save_csv() 

104 

105 selector_run = selector_scenario.selector.run_cli( 

106 scenario_path=selector_scenario.scenario_file, 

107 instance_set=data_set, 

108 feature_data=feature_dataframe.csv_filepath, 

109 run_on=run_on, 

110 slurm_prepend=settings.slurm_job_prepend, 

111 sbatch_options=settings.sbatch_settings, 

112 dependencies=feature_runs, 

113 log_dir=sl.caller_log_dir, 

114 ) 

115 

116 if run_on == Runner.LOCAL: 

117 selector_run.wait() 

118 print("Running Sparkle portfolio selector done!") 

119 else: 

120 print("Sparkle portfolio selector is running ...") 

121 

122 # Write used settings to file 

123 settings.write_used_settings() 

124 sys.exit(0) 

125 

126 

127if __name__ == "__main__": 

128 main(sys.argv[1:])