Coverage for src/sparkle/selector/selector_cli.py: 83%

1#!/usr/bin/env python3 

2# -*- coding: UTF-8 -*- 

3"""Execute Sparkle portfolio selector, read/write to SelectionScenario.""" 

4 

5import argparse 

6import sys 

7from filelock import FileLock, Timeout 

8from pathlib import Path 

9 

10from sparkle.structures import PerformanceDataFrame, FeatureDataFrame 

11from sparkle.solver import Solver 

12from sparkle.selector import SelectionScenario 

13from sparkle.instance import Instance_Set 

14 

15 

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

17 """Main function of the Selector CLI.""" 

18 # Define command line arguments 

19 parser = argparse.ArgumentParser() 

20 parser.add_argument( 

21 "--selector-scenario", 

22 required=True, 

23 type=Path, 

24 help="path to portfolio selector scenario", 

25 ) 

26 parser.add_argument( 

27 "--instance", required=True, type=Path, help="path to instance to run on" 

28 ) 

29 parser.add_argument( 

30 "--feature-data", required=True, type=Path, help="path to feature data" 

31 ) 

32 parser.add_argument( 

33 "--seed", 

34 type=int, 

35 required=False, 

36 help="seed to use for the solver. If not provided, read from " 

37 "the PerformanceDataFrame or generate one.", 

38 ) 

39 parser.add_argument( 

40 "--log-dir", type=Path, required=False, help="path to the log directory" 

41 ) 

42 args = parser.parse_args(argv) 

43 

44 # Process command line arguments 

45 selector_scenario = SelectionScenario.from_file(args.selector_scenario) 

46 feature_data = FeatureDataFrame(Path(args.feature_data)) 

47 instance_set = Instance_Set(args.instance) 

48 instance = instance_set.instance_paths[0] 

49 instance_name = instance_set.instance_names[0] 

50 instance_pair = (instance_set.directory.name, instance_name) 

51 seed = args.seed 

52 if seed is None: 

53 # Try to read from PerformanceDataFrame 

54 seed = selector_scenario.selector_performance_data.get_value( 

55 selector_scenario.__selector_solver_name__, 

56 instance_pair, 

57 solver_fields=[PerformanceDataFrame.column_seed], 

58 ) 

59 if seed is None: # Still no value 

60 import random 

61 

62 seed = random.randint(0, 2**32 - 1) 

63 

64 # Note: Following code could be adjusted to run entire instance set 

65 # Run portfolio selector 

66 print(f"Sparkle portfolio selector predicting for instance {instance_name} ...") 

67 if instance_pair not in feature_data.instance_pairs: 

68 raise ValueError( 

69 f"Could not resolve {instance} features in {feature_data.csv_filepath}" 

70 ) 

71 predict_schedule = selector_scenario.selector.run( 

72 selector_scenario.selector_file_path, 

73 instance_set.directory.name, 

74 instance_name, 

75 feature_data, 

76 ) 

77 

78 if predict_schedule is None: # Selector Failed to produce prediction 

79 sys.exit(-1) 

80 

81 print( 

82 f"Predicting done! Running schedule [{', '.join(str(x) for x in predict_schedule)}] ..." 

83 ) 

84 performance_data = selector_scenario.selector_performance_data 

85 selector_output = {} 

86 for solver, config_id, cutoff_time in predict_schedule: 

87 config = performance_data.get_full_configuration(solver, config_id) 

88 solver = Solver(Path(solver)) 

89 print( 

90 f"\t- Calling {solver.name} ({config_id}) with time budget {cutoff_time} " 

91 f"on instance {instance}..." 

92 ) 

93 solver_output = solver.run( # Runs locally by default 

94 instance_set, 

95 objectives=[selector_scenario.objective], 

96 seed=seed, 

97 cutoff_time=cutoff_time, 

98 configuration=config, 

99 log_dir=args.log_dir, 

100 ) 

101 for key in solver_output: 

102 if key in selector_output and isinstance(solver_output[key], (int, float)): 

103 selector_output[key] += solver_output[key] 

104 else: 

105 selector_output[key] = solver_output[key] 

106 print(f"\t- Calling solver {solver.name} ({config_id}) done!") 

107 solver_status = solver_output["status"] 

108 if solver_status.positive: 

109 print( 

110 f"[{solver_status}] {solver.name} ({config_id}) was succesfull on {instance}" 

111 ) 

112 break 

113 print(f"[{solver_status}] {solver.name} ({config_id}) failed on {instance}") 

114 

115 selector_value = selector_output[selector_scenario.objective.name] 

116 if selector_scenario.objective.post_process: 

117 selector_value = selector_scenario.objective.post_process( 

118 selector_output[selector_scenario.objective.name], 

119 cutoff_time, 

120 selector_output["status"], 

121 ) 

122 if solver_output["status"].positive: 

123 print( 

124 f"Selector {selector_scenario.selector.name} solved {instance} " 

125 f"with a value of {selector_value} ({selector_scenario.objective.name})." 

126 ) 

127 else: 

128 print(f"Selector {selector_scenario.selector.name} did not solve {instance}.") 

129 print(f"Writing results to {performance_data.csv_filepath} ...") 

130 try: 

131 # Creating a seperate locked file for writing 

132 lock = FileLock(f"{performance_data.csv_filepath}.lock") 

133 with lock.acquire(timeout=60): 

134 # Reload the dataframe to latest version 

135 performance_data = PerformanceDataFrame(performance_data.csv_filepath) 

136 performance_data.set_value( 

137 selector_value, 

138 selector_scenario.__selector_solver_name__, 

139 instance_pair, 

140 objective=selector_scenario.objective.name, 

141 append_write_csv=True, 

142 ) 

143 lock.release() 

144 except Timeout: 

145 print(f"ERROR: Cannot acquire File Lock on {performance_data}.") 

146 

147 

148if __name__ == "__main__": 

149 main(sys.argv[1:])