Coverage for src/sparkle/CLI/run_solvers.py: 88%
155 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-08 12:00 +0000
1#!/usr/bin/env python3
2"""Sparkle command to run solvers to get their performance data."""
4from __future__ import annotations
5import random
6import sys
7import argparse
8from pathlib import Path
10from runrunner.base import Runner, Run
12from sparkle.solver import Solver
13from sparkle.instance import Instance_Set
14from sparkle.structures import PerformanceDataFrame
15from sparkle.types import SparkleObjective, resolve_objective
16from sparkle.instance import InstanceSet
17from sparkle.platform.settings_objects import Settings
18from sparkle.CLI.help import global_variables as gv
19from sparkle.CLI.help import logging as sl
20from sparkle.CLI.help import argparse_custom as ac
21from sparkle.CLI.help.nicknames import resolve_object_name, resolve_instance_name
22from sparkle.CLI.initialise import check_for_initialise
25def parser_function() -> argparse.ArgumentParser:
26 """Define the command line arguments."""
27 parser = argparse.ArgumentParser(
28 description="Run solvers on instances to get their performance data."
29 )
30 parser.add_argument(*ac.SolversArgument.names, **ac.SolversArgument.kwargs)
31 parser.add_argument(
32 *ac.InstanceSetPathsArgument.names, **ac.InstanceSetPathsArgument.kwargs
33 )
35 # Mutually exclusive: specific configuration or best configuration
36 configuration_group = parser.add_mutually_exclusive_group()
37 configuration_group.add_argument(
38 *ac.ConfigurationArgument.names, **ac.ConfigurationArgument.kwargs
39 )
40 configuration_group.add_argument(
41 *ac.BestConfigurationArgument.names, **ac.BestConfigurationArgument.kwargs
42 )
43 configuration_group.add_argument(
44 *ac.AllConfigurationArgument.names, **ac.AllConfigurationArgument.kwargs
45 )
46 parser.add_argument(*ac.ObjectiveArgument.names, **ac.ObjectiveArgument.kwargs)
47 parser.add_argument(
48 *ac.PerformanceDataJobsArgument.names, **ac.PerformanceDataJobsArgument.kwargs
49 )
50 # This one is only relevant if the argument above is given
51 parser.add_argument(
52 *ac.RecomputeRunSolversArgument.names, **ac.RecomputeRunSolversArgument.kwargs
53 )
54 # Settings arguments
55 parser.add_argument(*ac.SettingsFileArgument.names, **ac.SettingsFileArgument.kwargs)
56 parser.add_argument(
57 *Settings.OPTION_solver_cutoff_time.args,
58 **Settings.OPTION_solver_cutoff_time.kwargs,
59 )
60 parser.add_argument(*Settings.OPTION_run_on.args, **Settings.OPTION_run_on.kwargs)
61 return parser
64def run_solvers(
65 solvers: list[Solver],
66 instances: list[str] | list[InstanceSet],
67 objectives: list[SparkleObjective],
68 seed: int,
69 cutoff_time: int,
70 configurations: list[list[dict[str, str]]],
71 sbatch_options: list[str] = None,
72 slurm_prepend: str | list[str] | Path = None,
73 log_dir: Path = None,
74 run_on: Runner = Runner.SLURM,
75) -> list[Run]:
76 """Run the solvers.
78 Parameters
79 ----------
80 solvers: list[solvers]
81 The solvers to run
82 instances: list[str] | list[InstanceSet]
83 The instances to run the solvers on
84 objectives: list[SparkleObjective]
85 The objective values to retrieve from the solvers
86 seed: int
87 The seed to use
88 cutoff_time: int
89 The cut off time for the solvers
90 configurations: list[list[str]]
91 The configurations to use for each solver
92 sbatch_options: list[str]
93 The sbatch options to use for the solvers
94 slurm_prepend: str | list[str] | Path
95 The script to prepend to a slurm script
96 log_dir: Path
97 The directory to use for the logs
98 run_on: Runner
99 Where to execute the solvers.
101 Returns
102 -------
103 run: runrunner.LocalRun or runrunner.SlurmRun
104 """
105 runs = []
106 # Run the solvers
107 for solver, solver_confs in zip(solvers, configurations):
108 for conf_index, conf in enumerate(solver_confs):
109 if "configuration_id" in conf.keys():
110 conf_name = conf["configuration_id"]
111 else:
112 conf_name = conf_index
113 run = solver.run(
114 instances=instances,
115 objectives=objectives,
116 seed=seed,
117 configuration=conf,
118 cutoff_time=cutoff_time,
119 run_on=run_on,
120 sbatch_options=sbatch_options,
121 slurm_prepend=slurm_prepend,
122 log_dir=log_dir,
123 )
124 if run_on == Runner.LOCAL:
125 if isinstance(run, dict):
126 run = [run]
127 # TODO: Refactor resolving objective keys
128 status_key = [key for key in run[0] if key.lower().startswith("status")][
129 0
130 ]
131 time_key = [key for key in run[0] if key.lower().startswith("cpu_time")][
132 0
133 ]
134 for i, solver_output in enumerate(run):
135 print(
136 f"Execution of {solver.name} ({conf_name}) on instance "
137 f"{instances[i]} completed with status "
138 f"{solver_output[status_key]} in {solver_output[time_key]} "
139 f"seconds."
140 )
141 print("Running configured solver done!")
142 else:
143 runs.append(run)
144 return runs
147def run_solvers_performance_data(
148 performance_data: PerformanceDataFrame,
149 cutoff_time: int,
150 rerun: bool = False,
151 solvers: list[Solver] = None,
152 instances: list[str] = None,
153 sbatch_options: list[str] = None,
154 slurm_prepend: str | list[str] | Path = None,
155 run_on: Runner = Runner.SLURM,
156) -> list[Run]:
157 """Run the solvers for the performance data.
159 Parameters
160 ----------
161 performance_data: PerformanceDataFrame
162 The performance data
163 cutoff_time: int
164 The cut off time for the solvers
165 rerun: bool
166 Run only solvers for which no data is available yet (False) or (re)run all
167 solvers to get (new) performance data for them (True)
168 solvers: list[solvers]
169 The solvers to run. If None, run all found solvers.
170 instances: list[str]
171 The instances to run the solvers on. If None, run all found instances.
172 sbatch_options: list[str]
173 The sbatch options to use
174 slurm_prepend: str | list[str] | Path
175 The script to prepend to a slurm script
176 run_on: Runner
177 Where to execute the solvers. For available values see runrunner.base.Runner
178 enum. Default: "Runner.SLURM".
180 Returns
181 -------
182 run: runrunner.LocalRun or runrunner.SlurmRun
183 If the run is local return a QueuedRun object with the information concerning
184 the run.
185 """
186 jobs = performance_data.remaining_jobs(rerun=rerun) # List of jobs to do
188 # Edit jobs to incorporate file paths
189 for index, (solver, config, (instance_set, instance_name), run) in enumerate(jobs):
190 instance_path = resolve_instance_name(
191 instance_set, instance_name, gv.settings().DEFAULT_instance_dir
192 )
193 jobs[index] = (solver, config, instance_path, run)
195 print(f"Total number of jobs to run: {len(jobs)}")
196 if len(jobs) == 0: # If there are no jobs, stop
197 return None
199 if run_on == Runner.LOCAL:
200 print("Running the solvers locally")
201 elif run_on == Runner.SLURM:
202 print("Running the solvers through Slurm")
204 if solvers is None:
205 solvers = [Solver(Path(s)) for s in performance_data.solvers]
206 else: # Filter the Solvers in remaining jobs
207 jobs = [
208 (solvers[solvers.index(solver)], configuration, instance, run)
209 for (solver, configuration, instance, run) in jobs
210 if solver in solvers
211 ]
213 if instances is not None: # Filter the instances
214 jobs = [job for job in jobs if job[2] in instances]
216 # Sort the jobs per solver
217 solver_jobs = {p_solver: {} for p_solver in solvers}
218 for p_solver, p_config, p_instance, p_run in jobs:
219 if p_config not in solver_jobs[p_solver]:
220 solver_jobs[p_solver][p_config] = {}
221 if p_instance not in solver_jobs[p_solver][p_config]:
222 solver_jobs[p_solver][p_config][p_instance] = [p_run]
223 else:
224 solver_jobs[p_solver][p_config][p_instance].append(p_run)
226 runrunner_runs = []
227 if run_on == Runner.LOCAL:
228 print(f"Cutoff time for each solver run: {cutoff_time} seconds")
229 for solver in solvers:
230 for solver_config in solver_jobs[solver].keys():
231 solver_instances = solver_jobs[solver][solver_config].keys()
232 run_ids = [
233 solver_jobs[solver][solver_config][instance]
234 for instance in solver_instances
235 ]
236 if solver_instances == []:
237 print(f"Warning: No jobs for instances found for solver {solver}")
238 continue
239 run = solver.run_performance_dataframe(
240 solver_instances,
241 performance_data,
242 solver_config,
243 run_ids=run_ids,
244 cutoff_time=cutoff_time,
245 sbatch_options=sbatch_options,
246 slurm_prepend=slurm_prepend,
247 log_dir=sl.caller_log_dir,
248 base_dir=sl.caller_log_dir,
249 run_on=run_on,
250 )
251 runrunner_runs.append(run)
252 if run_on == Runner.LOCAL:
253 # Do some printing?
254 pass
255 if run_on == Runner.SLURM:
256 num_jobs = sum(len(run.jobs) for run in runrunner_runs)
257 print(f"Total number of jobs submitted: {num_jobs}")
259 return runrunner_runs
262def main(argv: list[str]) -> None:
263 """Main function of the run solvers command."""
264 # Define command line arguments
265 parser = parser_function()
267 # Process command line arguments
268 args = parser.parse_args(argv)
269 settings = gv.settings(args)
271 # Log command call
272 sl.log_command(sys.argv, seed=settings.random_state)
273 check_for_initialise()
275 if args.best_configuration:
276 if not args.objective:
277 objective = settings.objectives[0]
278 print(
279 "WARNING: Best configuration requested, but no objective specified. "
280 f"Defaulting to first objective: {objective}"
281 )
282 else:
283 objective = resolve_objective(args.objective)
285 # Compare current settings to latest.ini
286 prev_settings = Settings(Settings.DEFAULT_previous_settings_path)
287 Settings.check_settings_changes(settings, prev_settings)
289 if args.solvers:
290 solvers = [
291 resolve_object_name(
292 solver_path,
293 gv.file_storage_data_mapping[gv.solver_nickname_list_path],
294 settings.DEFAULT_solver_dir,
295 Solver,
296 )
297 for solver_path in args.solvers
298 ]
299 else:
300 solvers = [
301 Solver(solver)
302 for solver in settings.DEFAULT_solver_dir.iterdir()
303 if solver.is_dir()
304 ]
306 if args.instance_path:
307 instances = [
308 resolve_object_name(
309 instance_path,
310 gv.file_storage_data_mapping[gv.instances_nickname_path],
311 settings.DEFAULT_instance_dir,
312 Instance_Set,
313 )
314 for instance_path in args.instance_path
315 ]
316 # Unpack the sets into instance strings
317 instances = [str(path) for set in instances for path in set.instance_paths]
318 else:
319 instances = None # TODO: Fix? Or its good like this
321 sbatch_options = settings.sbatch_settings
322 slurm_prepend = settings.slurm_job_prepend
323 # Write settings to file before starting, since they are used in callback scripts
324 settings.write_used_settings()
325 run_on = settings.run_on
326 cutoff_time = settings.solver_cutoff_time
327 # Open the performance data csv file
328 performance_dataframe = PerformanceDataFrame(settings.DEFAULT_performance_data_path)
330 print("Start running solvers ...")
331 if args.performance_data_jobs:
332 runs = run_solvers_performance_data(
333 performance_data=performance_dataframe,
334 solvers=solvers,
335 instances=instances,
336 cutoff_time=cutoff_time,
337 rerun=args.recompute,
338 sbatch_options=sbatch_options,
339 slurm_prepend=slurm_prepend,
340 run_on=run_on,
341 )
342 else:
343 if args.best_configuration:
344 train_instances = None
345 if isinstance(args.best_configuration, list):
346 train_instances = [
347 resolve_object_name(
348 instance_path,
349 gv.file_storage_data_mapping[gv.instances_nickname_path],
350 settings.DEFAULT_instance_dir,
351 Instance_Set,
352 )
353 for instance_path in args.best_configuration
354 ]
355 # Unpack the sets into instance strings
356 instances = [
357 str(path) for set in train_instances for path in set.instance_paths
358 ]
359 # Determine best configuration
360 configurations = [
361 [
362 performance_dataframe.best_configuration(
363 str(solver.directory), objective, train_instances
364 )[0]
365 ]
366 for solver in solvers
367 ]
368 elif args.configuration:
369 # Sort the configurations to the solvers
370 # TODO: Add a better check that the id could only match this solver
371 configurations = []
372 for solver in solvers:
373 configurations.append([])
374 for configuration in args.configuration:
375 if configuration not in performance_dataframe.configuration_ids:
376 raise ValueError(f"Configuration id {configuration} not found.")
377 if configuration in performance_dataframe.get_configurations(
378 str(solver.directory)
379 ):
380 configurations[-1].append(configuration)
381 elif args.all_configurations: # All known configurations
382 configurations = [
383 performance_dataframe.get_configurations(str(solver.directory))
384 for solver in solvers
385 ]
386 else: # Only default configurations
387 configurations = [
388 [PerformanceDataFrame.default_configuration] for _ in solvers
389 ]
390 # Look up and replace with the actual configurations
391 for solver_index, configs in enumerate(configurations):
392 for config_index, config in enumerate(configs):
393 configurations[solver_index][config_index] = (
394 performance_dataframe.get_full_configuration(
395 str(solvers[solver_index].directory), config
396 )
397 )
398 if instances is None:
399 instances = []
400 for instance_dir in settings.DEFAULT_instance_dir.iterdir():
401 if instance_dir.is_dir():
402 instances.append(Instance_Set(instance_dir))
404 # TODO Objective arg not used in Multi-file-instances case?
405 runs = run_solvers(
406 solvers=solvers,
407 configurations=configurations,
408 instances=instances,
409 objectives=settings.objectives,
410 seed=random.randint(0, 2**32 - 1),
411 cutoff_time=cutoff_time,
412 sbatch_options=sbatch_options,
413 slurm_prepend=slurm_prepend,
414 log_dir=sl.caller_log_dir,
415 run_on=run_on,
416 )
418 # If there are no jobs return
419 if runs is None or all(run is None for run in runs):
420 print("Running solvers done!")
421 elif run_on == Runner.SLURM:
422 print(
423 "Running solvers through Slurm with job id(s): "
424 f"{','.join(run.run_id for run in runs if run is not None)}"
425 )
426 sys.exit(0)
429if __name__ == "__main__":
430 main(sys.argv[1:])