Coverage for sparkle/solver/selector.py: 96%

1"""File to handle a Selector for selecting Solvers.""" 

2from __future__ import annotations 

3from pathlib import Path 

4 

5from sklearn.base import ClassifierMixin, RegressorMixin 

6from asf.cli import cli_train as asf_cli 

7from asf.scenario.scenario_metadata import ScenarioMetadata 

8from asf.predictors import AbstractPredictor 

9from asf.selectors.abstract_model_based_selector import AbstractModelBasedSelector 

10 

11import runrunner as rrr 

12from runrunner import Runner, Run 

13 

14from sparkle.types import SparkleObjective 

15from sparkle.structures import FeatureDataFrame, PerformanceDataFrame 

16 

17 

18class Selector: 

19 """The Selector class for handling Algorithm Selection.""" 

20 

21 def __init__( 

22 self: Selector, 

23 selector_class: AbstractModelBasedSelector, 

24 model_class: AbstractPredictor | ClassifierMixin | RegressorMixin) -> None: 

25 """Initialize the Selector object. 

26 

27 Args: 

28 selector_class: The Selector class to construct. 

29 model_class: The model class the selector will use. 

30 """ 

31 self.selector_class = selector_class 

32 self.model_class = model_class 

33 

34 @property 

35 def name(self: Selector) -> str: 

36 """Return the name of the selector.""" 

37 return f"{self.selector_class.__name__}_{self.model_class.__name__}" 

38 

39 def construct(self: Selector, 

40 target_file: Path, 

41 performance_data: PerformanceDataFrame, 

42 feature_data: FeatureDataFrame, 

43 objective: SparkleObjective, 

44 solver_cutoff: int | float | str = None, 

45 run_on: Runner = Runner.SLURM, 

46 sbatch_options: list[str] = None, 

47 slurm_prepend: str | list[str] | Path = None, 

48 base_dir: Path = Path()) -> Run: 

49 """Construct the Selector. 

50 

51 Args: 

52 target_file: Path to the file to save the Selector to. 

53 performance_data: Path to the performance data csv. 

54 feature_data: Path to the feature data csv. 

55 objective: The objective to optimize for selection. 

56 runtime_cutoff: Cutoff for the runtime in seconds. 

57 run_on: Which runner to use. Defaults to slurm. 

58 sbatch_options: Additional options to pass to sbatch. 

59 slurm_prepend: Slurm script to prepend to the sbatch 

60 base_dir: The base directory to run the Selector in. 

61 

62 Returns: 

63 The construction Run 

64 """ 

65 # Convert the dataframes to Selector Format 

66 # Requires instances as index for both, columns as features / solvers 

67 # Remove redundant data 

68 performance_csv = performance_data.drop( 

69 [PerformanceDataFrame.column_seed, 

70 PerformanceDataFrame.column_configuration], 

71 axis=1, level=1).droplevel(level=1, axis=1) 

72 performance_csv = performance_csv.loc[objective.name] # Select objective 

73 performance_csv.index = performance_csv.index.droplevel("Run") # Drop runs 

74 performance_path = target_file.parent / performance_data.csv_filepath.name 

75 performance_csv.to_csv(target_file.parent / performance_data.csv_filepath.name) 

76 

77 # Features requires instances as index, columns as feature names 

78 feature_csv = feature_data.dataframe.copy() 

79 feature_csv.index = feature_csv.index.map("_".join) # Reduce Multi-Index 

80 feature_csv = feature_csv.T # ASF has feature columns and instance rows 

81 feature_path = target_file.parent / feature_data.csv_filepath.name 

82 feature_csv.to_csv(feature_path) 

83 

84 selector = self.selector_class( 

85 self.model_class, ScenarioMetadata( 

86 algorithms=performance_data.solvers, 

87 features=feature_csv.columns.to_list(), 

88 performance_metric=objective.name, 

89 maximize=not objective.minimise, 

90 budget=solver_cutoff 

91 ) 

92 ) 

93 

94 cmd = asf_cli.build_cli_command(selector, 

95 feature_path, 

96 performance_path, 

97 target_file) 

98 

99 cmd = [" ".join([str(c) for c in cmd])] 

100 construct = rrr.add_to_queue( 

101 runner=run_on, 

102 cmd=cmd, 

103 name=f"{self.name} Selector Construction: " 

104 f"{', '.join([Path(s).name for s in performance_data.solvers])}", 

105 base_dir=base_dir, 

106 sbatch_options=sbatch_options, 

107 prepend=slurm_prepend) 

108 if run_on == Runner.LOCAL: 

109 construct.wait() 

110 if not target_file.is_file(): 

111 print(f"Selector construction of {self.name} failed!") 

112 

113 return construct 

114 

115 def run(self: Selector, 

116 selector_path: Path, 

117 instance: str, 

118 feature_data: FeatureDataFrame) -> list: 

119 """Run the Selector, returning the prediction schedule upon success.""" 

120 instance_features = feature_data.dataframe[[instance, ]] 

121 instance_features.index = instance_features.index.map("_".join) # Reduce 

122 instance_features = instance_features.T # ASF dataframe structure 

123 selector = self.selector_class.load(selector_path) 

124 schedule = selector.predict(instance_features) 

125 if schedule is None: 

126 print(f"ERROR: Selector {self.name} failed predict schedule!") 

127 return schedule[instance]