Coverage for src/sparkle/selector/selector.py: 87%
160 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-08 12:00 +0000
1"""File to handle a Selector for selecting Solvers."""
3from __future__ import annotations
4import random
5from pathlib import Path
8from sklearn.base import ClassifierMixin, RegressorMixin
9from asf.cli import cli_train as asf_cli
10from asf.predictors import AbstractPredictor
11from asf.selectors.abstract_model_based_selector import AbstractModelBasedSelector
14import runrunner as rrr
15from runrunner import Runner, Run
16import pandas as pd
18from sparkle.types import SparkleObjective, resolve_objective
19from sparkle.structures import FeatureDataFrame, PerformanceDataFrame
20from sparkle.instance import InstanceSet
23class Selector:
24 """The Selector class for handling Algorithm Selection."""
26 selector_cli = Path(__file__).parent / "selector_cli.py"
28 def __init__(
29 self: Selector,
30 selector_class: AbstractModelBasedSelector,
31 model_class: AbstractPredictor | ClassifierMixin | RegressorMixin,
32 ) -> None:
33 """Initialize the Selector object.
35 Args:
36 selector_class: The (name of) Selector class to construct.
37 model_class: The (name of) model class the selector will use.
38 """
39 if isinstance(selector_class, str): # Resolve class name
40 from asf import selectors
42 selector_class = getattr(selectors, selector_class)
43 if isinstance(model_class, str): # Resolve class name
44 from sklearn import ensemble
46 model_class = getattr(ensemble, model_class)
48 self.selector_class = selector_class
49 self.model_class = model_class
51 @property
52 def name(self: Selector) -> str:
53 """Return the name of the selector."""
54 return f"{self.selector_class.__name__}_{self.model_class.__name__}"
56 def construct(
57 self: Selector,
58 selection_scenario: SelectionScenario,
59 run_on: Runner = Runner.SLURM,
60 job_name: str = None,
61 sbatch_options: list[str] = None,
62 slurm_prepend: str | list[str] | Path = None,
63 base_dir: Path = Path(),
64 ) -> Run:
65 """Construct the Selector.
67 Args:
68 selection_scenario: The scenario to construct the Selector for.
69 run_on: Which runner to use. Defaults to slurm.
70 job_name: Name to give the construction job when submitting.
71 sbatch_options: Additional options to pass to sbatch.
72 slurm_prepend: Slurm script to prepend to the sbatch
73 base_dir: The base directory to run the Selector in.
75 Returns:
76 The construction Run
77 """
78 selection_scenario.create_scenario()
79 selector = self.selector_class(
80 model_class=self.model_class,
81 budget=selection_scenario.solver_cutoff,
82 maximize=not selection_scenario.objective.minimise,
83 )
84 cmd = asf_cli.build_cli_command(
85 selector,
86 selection_scenario.feature_target_path,
87 selection_scenario.performance_target_path,
88 selection_scenario.selector_file_path,
89 )
90 cmd = [" ".join([str(c) for c in cmd])]
92 job_name = job_name or f"Selector Construction {selection_scenario.name}"
93 construct = rrr.add_to_queue(
94 runner=run_on,
95 cmd=cmd,
96 name=job_name,
97 base_dir=base_dir,
98 sbatch_options=sbatch_options,
99 prepend=slurm_prepend,
100 )
102 if run_on == Runner.LOCAL:
103 construct.wait()
104 if not selection_scenario.selector_file_path.is_file():
105 print(f"Selector construction of {self.name} failed!")
106 return construct
108 def run(
109 self: Selector,
110 selector_path: Path,
111 instance_set: str,
112 instance_name: str,
113 feature_data: FeatureDataFrame,
114 ) -> list:
115 """Run the Selector, returning the prediction schedule upon success.
117 Args:
118 selector_path: The path to the selector to run.
119 instance_set: The name of the set the instance belongs to. Required because
120 the feature data is keyed on the (set, instance) pair, so a bare instance
121 name (or file path) could resolve to the wrong set.
122 instance_name: The name of the instance to predict for.
123 feature_data: The instance feature data to use.
124 """
125 instance_features = feature_data.get_instance(
126 instance_set, instance_name, as_dataframe=True
127 )
128 # ASF was trained on plain instance names, drop InstanceSet level
129 instance_features = instance_features.droplevel(
130 FeatureDataFrame.instance_set_index_dim
131 )
132 selector = self.selector_class.load(selector_path)
133 schedule = selector.predict(instance_features)
134 if schedule is None:
135 print(f"ERROR: Selector {self.name} failed predict schedule!")
136 return None
137 # ASF presents result as schedule per instance, we only use one in this setting
138 schedule = schedule[instance_name]
139 for index, (solver, time) in enumerate(schedule):
140 # Split solver name back into solver and config id
141 # NOTE: There is an issue with this incase the Solver name has an "_" in its name... We need to change the delimiter to different character(s)
142 solver_name, conf_index = solver.split("_", maxsplit=1)
143 schedule[index] = (solver_name, conf_index, time)
144 return schedule
146 def run_cli(
147 self: Selector,
148 scenario_path: Path,
149 instance_set: InstanceSet | list[Path],
150 feature_data: Path,
151 run_on: Runner = Runner.LOCAL,
152 sbatch_options: list[str] = None,
153 slurm_prepend: str | list[str] | Path = None,
154 job_name: str = None,
155 dependencies: list[Run] = None,
156 log_dir: Path = None,
157 ) -> Run:
158 """Run the Selector CLI and write result to the Scenario PerformanceDataFrame.
160 Args:
161 scenario_path: The path to the scenario with the Selector to run.
162 instance_set: The instance set to run the Selector on.
163 feature_data: The instance feature data to use.
164 run_on: Which runner to use. Defaults to slurm.
165 sbatch_options: Additional options to pass to sbatch.
166 slurm_prepend: Slurm script to prepend to the sbatch
167 job_name: Name to give the Slurm job when submitting.
168 dependencies: List of dependencies to add to the job.
169 log_dir: The directory to write logs to.
171 Returns:
172 The Run object.
173 """
174 # NOTE: The selector object and the scenario selector could differ which could
175 # cause unintended behaviour (e.g. running a different selector than desired)
176 instances = (
177 instance_set if isinstance(instance_set, list) else instance_set.instances
178 )
179 commands = [
180 f"python3 {Selector.selector_cli} "
181 f"--selector-scenario {scenario_path} "
182 f"--instance {instance} "
183 f"--feature-data {feature_data} "
184 f"--log-dir {log_dir} "
185 f"--seed {random.randint(0, 2**32 - 1)}"
186 for instance in instances
187 ]
189 job_name = (
190 f"Run Selector {self.name} on {len(instances)} instances"
191 if not job_name
192 else job_name
193 )
194 import subprocess
196 r = rrr.add_to_queue(
197 cmd=commands,
198 name=job_name,
199 stdout=None if run_on == Runner.LOCAL else subprocess.PIPE, # Print
200 stderr=None if run_on == Runner.LOCAL else subprocess.PIPE, # Print
201 base_dir=log_dir,
202 runner=run_on,
203 sbatch_options=sbatch_options,
204 prepend=slurm_prepend,
205 dependencies=dependencies,
206 )
207 if run_on == Runner.LOCAL:
208 r.wait()
209 return r
212class SelectionScenario:
213 """A scenario for a Selector."""
215 __selector_solver_name__ = "portfolio_selector"
217 def __init__(
218 self: SelectionScenario,
219 parent_directory: Path,
220 selector: Selector,
221 objective: SparkleObjective,
222 performance_data: PerformanceDataFrame | Path,
223 feature_data: FeatureDataFrame | Path,
224 feature_extractors: list[str] = None,
225 solver_cutoff: int | float = None,
226 extractor_cutoff: int | float = None,
227 ablate: bool = False,
228 subdir_path: Path = None,
229 ) -> None:
230 """Initialize a scenario for a selector."""
231 self.selector: Selector = selector
232 self.objective: SparkleObjective = objective
233 self.solver_cutoff: float = solver_cutoff
234 self.extractor_cutoff: float = extractor_cutoff
235 if subdir_path is not None:
236 self.directory = parent_directory / subdir_path
237 elif isinstance(performance_data, PerformanceDataFrame):
238 self.directory: Path = (
239 parent_directory
240 / selector.name
241 / "_".join([Path(s).name for s in performance_data.solvers])
242 )
243 else:
244 self.directory = performance_data.parent
245 self.name = f"{selector.name} on {self.directory.name}"
246 self.selector_file_path: Path = self.directory / "portfolio_selector"
247 self.scenario_file: Path = self.directory / "scenario.txt"
248 self.selector_performance_path: Path = (
249 self.directory / "selector_performance.csv"
250 )
251 if self.selector_performance_path.exists():
252 self.selector_performance_data = PerformanceDataFrame(
253 self.selector_performance_path
254 )
255 else: # Create new performance data frame for selector, write to file later
256 self.selector_performance_data = performance_data.clone()
257 self.selector_performance_data.add_solver(
258 SelectionScenario.__selector_solver_name__
259 )
261 if isinstance(performance_data, PerformanceDataFrame): # Convert
262 # Store (InstanceSet, Instance) pairs before collapsing index for ASF
263 self.training_instance_pairs: list[tuple[str, str]] = list(
264 dict.fromkeys(
265 zip(
266 performance_data.index.get_level_values(
267 PerformanceDataFrame.index_instance_set
268 ),
269 performance_data.index.get_level_values(
270 PerformanceDataFrame.index_instance
271 ),
272 )
273 )
274 )
275 # Convert the dataframes to Selector Format
276 new_column_names: list[str] = []
277 for solver, config_id, _ in performance_data.columns:
278 if f"{solver}_{config_id}" not in new_column_names:
279 new_column_names.append(f"{solver}_{config_id}")
280 self.performance_data = performance_data.drop(
281 [PerformanceDataFrame.column_seed], axis=1, level=2
282 )
283 self.performance_data = self.performance_data.droplevel(
284 [
285 PerformanceDataFrame.column_configuration,
286 PerformanceDataFrame.column_meta,
287 ],
288 axis=1,
289 )
290 self.performance_data = self.performance_data.droplevel(
291 PerformanceDataFrame.index_objective, axis=0
292 )
293 self.performance_data.columns = new_column_names
294 # Requires instances as index for both, columns as features / solvers
295 # Drop InstanceSet and Run levels so ASF sees plain instance names
296 self.performance_data.index = self.performance_data.index.droplevel(
297 [PerformanceDataFrame.index_instance_set, PerformanceDataFrame.index_run]
298 )
299 # Enforce data type to be numeric
300 self.performance_data = self.performance_data.astype(float)
301 self.performance_target_path = self.directory / "performance_data.csv"
302 else: # Read from Path
303 self.performance_data: pd.DataFrame = pd.read_csv(
304 performance_data, index_col=0
305 )
306 self.performance_target_path: Path = performance_data
307 # The flat performance CSV is indexed by instance name only; recover the
308 # (set, instance) training pairs by matching those names against the selector
309 # performance data, which keeps the full pair index. Without this, a scenario
310 # loaded from file would report zero training instances.
311 training_names = set(self.performance_data.index)
312 self.training_instance_pairs: list[tuple[str, str]] = [
313 (instance_set, instance_name)
314 for instance_set, instance_name in self.selector_performance_data.instance_pairs
315 if instance_name in training_names
316 ]
318 if isinstance(feature_data, FeatureDataFrame): # Convert
319 self.feature_extractors = feature_data.extractors
320 # Features requires instances as index, columns as feature names
321 feature_target = feature_data.copy()
322 feature_target.columns = feature_target.columns.map(
323 "_".join
324 ) # Reduce Column Multi Index to single
325 # Drop InstanceSet level so ASF sees plain instance names
326 feature_target.index = feature_target.index.droplevel(
327 FeatureDataFrame.instance_set_index_dim
328 )
329 # ASF -> feature columns, instance rows
330 self.feature_data: pd.DataFrame = feature_target.astype(float)
331 self.feature_target_path: Path = self.directory / "feature_data.csv"
332 else: # Read from Path
333 self.feature_extractors = feature_extractors
334 self.feature_data: pd.DataFrame = pd.read_csv(feature_data)
335 self.feature_target_path: Path = feature_data
337 self.ablation_scenarios: list[SelectionScenario] = []
338 if ablate and len(self.performance_data.columns) > 2:
339 for solver in self.performance_data.columns:
340 solver_key, conf_id = solver.split("_", maxsplit=1)
341 ablate_subdir = Path(f"ablated_{Path(solver).name}")
342 ablated_directory = self.directory / ablate_subdir
343 if (ablated_directory / "performance_data.csv").exists():
344 ablated_pd = ablated_directory / "performance_data.csv"
345 elif isinstance(performance_data, PerformanceDataFrame):
346 ablated_pd = performance_data.clone()
347 ablated_pd.remove_configuration(solver_key, conf_id)
348 else: # Note we could do this but it would be hacky?
349 raise ValueError(
350 "Cannot ablate scenario after loading from file! "
351 "Requires original PerformanceDataFrame."
352 )
354 self.ablation_scenarios.append(
355 SelectionScenario(
356 parent_directory=self.directory,
357 selector=selector,
358 objective=objective,
359 performance_data=ablated_pd,
360 feature_data=feature_data,
361 solver_cutoff=solver_cutoff,
362 ablate=False, # If we set to true here, recursion would happen
363 subdir_path=ablate_subdir,
364 )
365 )
367 @property
368 def training_instances(self: SelectionScenario) -> list[tuple[str, str]]:
369 """Get the training instances as (set_name, instance_name) pairs."""
370 return self.training_instance_pairs
372 @property
373 def test_instances(self: SelectionScenario) -> list[tuple[str, str]]:
374 """Get the test instances as (set_name, instance_name) pairs."""
375 training_set = set(self.training_instances)
376 return [
377 instance_pair
378 for instance_pair in self.selector_performance_data.instance_pairs
379 if instance_pair not in training_set
380 ]
382 @property
383 def training_instance_sets(self: SelectionScenario) -> list[str]:
384 """Get the training instance sets."""
385 return list(
386 dict.fromkeys(instance_set for instance_set, _ in self.training_instances)
387 )
389 @property
390 def test_instance_sets(self: SelectionScenario) -> list[str]:
391 """Get the test instance sets."""
392 return list(
393 dict.fromkeys(instance_set for instance_set, _ in self.test_instances)
394 )
396 @property
397 def instance_sets(self: SelectionScenario) -> list[str]:
398 """Get all the instance sets used in this scenario."""
399 return list(
400 dict.fromkeys(
401 instance_set
402 for instance_set, _ in self.selector_performance_data.instance_pairs
403 )
404 )
406 @property
407 def solvers(self: SelectionScenario) -> list[str]:
408 """Get the solvers used for the selector."""
409 return self.performance_data.columns.to_list()
411 def create_scenario(self: SelectionScenario) -> None:
412 """Prepare the scenario directories."""
413 self.directory.mkdir(parents=True, exist_ok=True)
414 self.performance_data.to_csv(self.performance_target_path)
415 self.feature_data.to_csv(self.feature_target_path)
416 self.selector_performance_data.save_csv(self.selector_performance_path)
417 self.create_scenario_file()
419 def create_scenario_file(self: SelectionScenario) -> None:
420 """Create the scenario file.
422 Write the scenario to file.
423 """
424 with self.scenario_file.open("w") as fout:
425 fout.write(self.serialise())
427 def serialise(self: SelectionScenario) -> dict:
428 """Serialize the scenario."""
429 return (
430 f"selector: {self.selector.name}\n"
431 f"solver_cutoff: {self.solver_cutoff}\n"
432 f"extractor_cutoff: {self.extractor_cutoff}\n"
433 f"ablate: {len(self.ablation_scenarios) > 0}\n"
434 f"objective: {self.objective}\n"
435 f"selector_performance_data: {self.selector_performance_path}\n"
436 f"performance_data: {self.performance_target_path}\n"
437 f"feature_data: {self.feature_target_path}\n"
438 f"feature_extractors: {','.join(self.feature_extractors)}\n"
439 )
441 @staticmethod
442 def from_file(scenario_file: Path) -> SelectionScenario:
443 """Reads scenario file and initalises SelectorScenario."""
444 if not scenario_file.is_file() and (scenario_file / "scenario.txt").is_file():
445 scenario_file = scenario_file / "scenario.txt" # Resolve from directory
446 values = {
447 key: value.strip()
448 for key, value in [
449 line.split(": ", maxsplit=1) for line in scenario_file.open()
450 ]
451 }
452 selector_class, selector_model = values["selector"].split("_", maxsplit=1)
453 import ast
455 selector = Selector(selector_class, selector_model)
456 return SelectionScenario(
457 parent_directory=scenario_file.parent,
458 selector=selector,
459 objective=resolve_objective(values["objective"]),
460 performance_data=Path(values["performance_data"]),
461 feature_data=Path(values["feature_data"]),
462 feature_extractors=values["feature_extractors"].split(","),
463 solver_cutoff=float(values["solver_cutoff"]),
464 extractor_cutoff=float(values["extractor_cutoff"]),
465 ablate=ast.literal_eval(values["ablate"]),
466 )