Coverage for src/sparkle/CLI/compute_features.py: 86%

1#!/usr/bin/env python3 

2"""Sparkle command to compute features for instances.""" 

3 

4from __future__ import annotations 

5import sys 

6import argparse 

7 

8from pathlib import Path 

9from runrunner.base import Run, Runner 

10 

11from sparkle.selector import Extractor 

12from sparkle.platform.settings_objects import Settings 

13from sparkle.structures import FeatureDataFrame 

14from sparkle.instance import Instance_Set, InstanceSet 

15 

16 

17from sparkle.CLI.help import global_variables as gv 

18from sparkle.CLI.help import logging as sl 

19from sparkle.CLI.help import argparse_custom as ac 

20from sparkle.CLI.initialise import check_for_initialise 

21from sparkle.CLI.help.nicknames import resolve_object_name, resolve_instance_name 

22 

23 

24def parser_function() -> argparse.ArgumentParser: 

25 """Define the command line arguments.""" 

26 parser = argparse.ArgumentParser( 

27 description="Sparkle command to Compute features " 

28 "for instances using added extractors " 

29 "and instances." 

30 ) 

31 parser.add_argument( 

32 *ac.InstanceSetPathsArgument.names, **ac.InstanceSetPathsArgument.kwargs 

33 ) 

34 parser.add_argument(*ac.ExtractorsArgument.names, **ac.ExtractorsArgument.kwargs) 

35 parser.add_argument( 

36 *ac.RecomputeFeaturesArgument.names, **ac.RecomputeFeaturesArgument.kwargs 

37 ) 

38 # Settings arguments 

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

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

41 parser.add_argument( 

42 *Settings.OPTION_groupwise_computation.args, 

43 **Settings.OPTION_groupwise_computation.kwargs, 

44 ) 

45 return parser 

46 

47 

48def compute_features( 

49 feature_data: FeatureDataFrame, 

50 recompute: bool, 

51 run_on: Runner = Runner.SLURM, 

52 instance_sets: list[InstanceSet] = None, 

53) -> list[Run]: 

54 """Compute features for all instance and feature extractor combinations. 

55 

56 A RunRunner run is submitted for the computation of the features. 

57 The results are then stored in the csv file specified by feature_data_csv_path. 

58 

59 Args: 

60 feature_data: Feature Data Frame to use 

61 recompute: Specifies if features should be recomputed. 

62 run_on: Runner 

63 On which computer or cluster environment to run the solvers. 

64 Available: Runner.LOCAL, Runner.SLURM. Default: Runner.SLURM 

65 instance_sets: Optional resolved instance sets to search. By default, instances 

66 are resolved from the registered platform instance directory. 

67 

68 Returns: 

69 Submitted runs. Empty if there are no jobs to execute. 

70 """ 

71 settings = gv.settings() 

72 if recompute: 

73 feature_data.reset_dataframe() 

74 

75 remaining_jobs = feature_data.remaining_jobs( 

76 groupwise_computation=settings.groupwise_computation 

77 ) 

78 

79 # If there are no jobs, stop 

80 if not remaining_jobs: 

81 print( 

82 "No feature computation jobs to run; stopping execution! To recompute " 

83 "feature values use the --recompute flag." 

84 ) 

85 return [] 

86 

87 cutoff = settings.extractor_cutoff_time 

88 sbatch_options = settings.sbatch_settings 

89 slurm_prepend = settings.slurm_job_prepend 

90 srun_options = ["-N1", "-n1"] + sbatch_options 

91 search_location = ( 

92 settings.DEFAULT_instance_dir if instance_sets is None else instance_sets 

93 ) 

94 runs = [] 

95 for (instance_set, instance_name), extractor_name, feature_group in remaining_jobs: 

96 extractor_path = settings.DEFAULT_extractor_dir / extractor_name 

97 extractor = Extractor(extractor_path) 

98 

99 instance_path = resolve_instance_name( 

100 instance_set, instance_name, search_location 

101 ) 

102 if instance_path is None: 

103 raise ValueError( 

104 f"ERROR: The instance {instance_name} ({instance_set}) could not be found. " 

105 f"Please make sure the path is correct." 

106 ) 

107 

108 instance_paths = [] 

109 if isinstance(instance_path, list): 

110 instance_paths = [Path(path) for path in instance_path] 

111 elif isinstance(instance_path, (str, Path)): 

112 instance_paths = [Path(instance_path)] 

113 

114 run = extractor.run_cli( 

115 instance_paths, 

116 feature_data, 

117 cutoff, 

118 feature_group, 

119 run_on, 

120 sbatch_options, 

121 srun_options, 

122 settings.slurm_jobs_in_parallel, 

123 slurm_prepend, 

124 log_dir=sl.caller_log_dir, 

125 ) 

126 runs.append(run) 

127 return runs 

128 

129 

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

131 """Main function of the compute features command.""" 

132 # Define command line arguments 

133 parser = parser_function() 

134 

135 # Process command line arguments 

136 args = parser.parse_args(argv) 

137 settings = gv.settings(args) 

138 run_on = settings.run_on 

139 

140 # Log command call 

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

142 check_for_initialise() 

143 

144 # Check if there are any feature extractors registered 

145 if not any([p.is_dir() for p in gv.settings().DEFAULT_extractor_dir.iterdir()]): 

146 print( 

147 "No feature extractors present! Add feature extractors to Sparkle " 

148 "by using the add_feature_extractor command." 

149 ) 

150 sys.exit() 

151 

152 # Load feature data 

153 feature_data = FeatureDataFrame(settings.DEFAULT_feature_data_path) 

154 

155 # Narrow the work down to only the instances and/or extractors the user named on CLI by filtering the full FDF. 

156 if args.instance_path: 

157 instances = set() 

158 for instance_arg in args.instance_path: 

159 instance: InstanceSet = resolve_object_name( 

160 instance_arg, 

161 gv.instance_set_nickname_mapping, 

162 settings.DEFAULT_instance_dir, 

163 Instance_Set, 

164 ) 

165 if instance is None: 

166 raise ValueError( 

167 f"Argument Error! Could not resolve instance: '{instance_arg}'" 

168 ) 

169 # resolve_object_name may hand back a single-file FileInstanceSet whose .name 

170 # is the file stem rather than the owning set (e.g. "PTN/bce7824.cnf" -> name 

171 # "bce7824"). The FeatureDataFrame is keyed by the owning set, so pair each 

172 # requested instance with its directory name to match how it was stored. 

173 for instance_name in instance.instance_names: 

174 instances.add((instance.directory.name, instance_name)) 

175 

176 filtered_instances = [ 

177 (instance_set, instance_name) 

178 for instance_set, instance_name in feature_data.instance_pairs 

179 if (instance_set, instance_name) not in instances 

180 ] 

181 feature_data.remove_instance(filtered_instances) 

182 if feature_data.num_instances == 0: 

183 raise ValueError("Argument Error! No instances left after filtering.") 

184 if args.extractors: 

185 extractors = set() 

186 for extractor in args.extractors: 

187 extractor: Extractor = resolve_object_name( 

188 extractor, 

189 nickname_dict=gv.extractor_nickname_mapping, 

190 target_dir=settings.DEFAULT_extractor_dir, 

191 class_name=Extractor, 

192 ) 

193 if extractor is None: 

194 raise ValueError( 

195 f"Argument Error! Could not resolve extractor: '{extractor}'" 

196 ) 

197 extractors.add(extractor.name) 

198 for extractor in feature_data.extractors: 

199 if extractor not in extractors: 

200 feature_data.remove_extractor(extractor) 

201 if feature_data.num_extractors == 0: 

202 raise ValueError( 

203 "Argument Error! No feature extractors left after filtering." 

204 ) 

205 

206 # Start compute features 

207 print("Start computing features ...") 

208 compute_features(feature_data, args.recompute, run_on) 

209 

210 # Write used settings to file 

211 gv.settings().write_used_settings() 

212 sys.exit(0) 

213 

214 

215if __name__ == "__main__": 

216 main(sys.argv[1:])