Coverage for sparkle/CLI/run_configured_solver.py: 89%

1#!/usr/bin/env python3 

2"""Sparkle command to execute a configured solver.""" 

3 

4import sys 

5import argparse 

6from pathlib import PurePath 

7 

8from runrunner.base import Runner 

9 

10from sparkle.CLI.help import global_variables as gv 

11from sparkle.CLI.help import logging as sl 

12from sparkle.platform.settings_objects import Settings, SettingState 

13from sparkle.instance import Instance_Set 

14from sparkle.platform import CommandName, COMMAND_DEPENDENCIES 

15from sparkle.CLI.initialise import check_for_initialise 

16from sparkle.CLI.help import argparse_custom as ac 

17from sparkle.CLI.help.nicknames import resolve_object_name 

18 

19 

20def parser_function() -> argparse.ArgumentParser: 

21 """Define the command line arguments.""" 

22 parser = argparse.ArgumentParser( 

23 description="Command to run a configured solver on an instance (set).") 

24 parser.add_argument(*ac.InstancePathPositional.names, 

25 **ac.InstancePathPositional.kwargs) 

26 parser.add_argument(*ac.SettingsFileArgument.names, 

27 **ac.SettingsFileArgument.kwargs) 

28 parser.add_argument(*ac.SparkleObjectiveArgument.names, 

29 **ac.SparkleObjectiveArgument.kwargs) 

30 parser.add_argument(*ac.RunOnArgument.names, 

31 **ac.RunOnArgument.kwargs) 

32 return parser 

33 

34 

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

36 """Main function of the run configured solver command.""" 

37 # Log command call 

38 sl.log_command(sys.argv) 

39 

40 # Define command line arguments 

41 parser = parser_function() 

42 

43 # Process command line arguments 

44 args = parser.parse_args(argv) 

45 

46 check_for_initialise(COMMAND_DEPENDENCIES[CommandName.RUN_CONFIGURED_SOLVER]) 

47 

48 if args.settings_file is not None: 

49 # Do first, so other command line options can override settings from the file 

50 gv.settings().read_settings_ini(args.settings_file, SettingState.CMD_LINE) 

51 if args.objectives is not None: 

52 gv.settings().set_general_sparkle_objectives( 

53 args.objectives, SettingState.CMD_LINE 

54 ) 

55 if args.run_on is not None: 

56 gv.settings().set_run_on( 

57 args.run_on.value, SettingState.CMD_LINE) 

58 

59 # Compare current settings to latest.ini 

60 prev_settings = Settings(PurePath("Settings/latest.ini")) 

61 Settings.check_settings_changes(gv.settings(), prev_settings) 

62 

63 # Try to resolve the instance path (Dir or list instance paths) 

64 data_set = resolve_object_name( 

65 args.instance_path, 

66 gv.file_storage_data_mapping[gv.instances_nickname_path], 

67 gv.settings().DEFAULT_instance_dir, Instance_Set) 

68 if data_set is None: 

69 print(f"Could not resolve instance (set): {args.instance_path}! Exiting...") 

70 sys.exit(-1) 

71 

72 run_on = gv.settings().get_run_on() 

73 # Get the name of the configured solver and the training set 

74 solver = gv.latest_scenario().get_config_solver() 

75 train_set = gv.latest_scenario().get_config_instance_set_train() 

76 custom_cutoff = gv.settings().get_general_target_cutoff_time() 

77 if solver is None or train_set is None: 

78 # Print error and stop execution 

79 print("ERROR: No configured solver found! Stopping execution.") 

80 sys.exit(-1) 

81 # Get optimised configuration 

82 configurator = gv.settings().get_general_sparkle_configurator() 

83 objectives = gv.settings().get_general_sparkle_objectives() 

84 configuration_scenario = gv.latest_scenario().get_configuration_scenario( 

85 configurator.scenario_class) 

86 _, config_str = configurator.get_optimal_configuration(configuration_scenario) 

87 config = solver.config_str_to_dict(config_str) 

88 # Call the configured solver 

89 sbatch_options = gv.settings().get_slurm_extra_options(as_args=True) 

90 if run_on == Runner.LOCAL: 

91 print(f"Start running solver on {data_set.size} instances...") 

92 run = solver.run(instance=data_set, 

93 objectives=objectives, 

94 seed=gv.get_seed(), 

95 cutoff_time=custom_cutoff, 

96 configuration=config, 

97 run_on=run_on, 

98 commandname=CommandName.RUN_CONFIGURED_SOLVER, 

99 sbatch_options=sbatch_options, 

100 log_dir=sl.caller_log_dir) 

101 

102 # Print result 

103 if run_on == Runner.SLURM: 

104 print(f"Running configured solver. Waiting for Slurm " 

105 f"job(s) with id(s): {run.run_id}") 

106 else: 

107 if isinstance(run, dict): 

108 run = [run] 

109 for i, solver_output in enumerate(run): 

110 print(f"Execution of {solver.name} on instance " 

111 f"{data_set.instance_names[i]} completed with status " 

112 f"{solver_output['status']} in {solver_output['cpu_time']} seconds.") 

113 print("Running configured solver done!") 

114 

115 # Write used settings to file 

116 gv.settings().write_used_settings() 

117 sys.exit(0) 

118 

119 

120if __name__ == "__main__": 

121 main(sys.argv[1:])