Coverage for src/sparkle/solver/solver.py: 91%
235 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-08 12:00 +0000
1"""File to handle a solver and its directories."""
3from __future__ import annotations
4import sys
5from typing import Any
6import shlex
7import ast
8import json
9import random
10from pathlib import Path
12from ConfigSpace import ConfigurationSpace
14import runrunner as rrr
15from runrunner.local import LocalRun
16from runrunner.slurm import Run, SlurmRun
17from runrunner.base import Status, Runner
19from sparkle.tools.parameters import PCSConverter, PCSConvention
20from sparkle.tools import RunSolver
21from sparkle.types import SparkleCallable, SolverStatus
22from sparkle.solver import verifiers
23from sparkle.instance import InstanceSet
24from sparkle.structures import PerformanceDataFrame
25from sparkle.types import resolve_objective, SparkleObjective, UseTime
28class Solver(SparkleCallable):
29 """Class to handle a solver and its directories."""
31 meta_data = "solver_meta.txt"
32 _wrapper_file = "sparkle_solver_wrapper"
33 solver_cli = Path(__file__).parent / "solver_cli.py"
35 def __init__(
36 self: Solver,
37 directory: Path,
38 runsolver_exec: Path = None,
39 deterministic: bool = None,
40 verifier: verifiers.SolutionVerifier = None,
41 ) -> None:
42 """Initialize solver.
44 Args:
45 directory: Directory of the solver.
46 runsolver_exec: Path to the runsolver executable.
47 By default, runsolver in directory.
48 deterministic: Bool indicating determinism of the algorithm.
49 Defaults to False.
50 verifier: The solution verifier to use. If None, no verifier is used.
51 """
52 super().__init__(directory, runsolver_exec)
53 self.deterministic = deterministic
54 self.verifier = verifier
55 self._pcs_file: Path = None
56 self._interpreter: str = None
57 self._wrapper_extension: str = None
59 meta_data_file = self.directory / Solver.meta_data
60 if meta_data_file.exists():
61 meta_data = ast.literal_eval(meta_data_file.open().read())
62 # We only override the deterministic and verifier from file if not set
63 if self.deterministic is None:
64 if (
65 "deterministic" in meta_data
66 and meta_data["deterministic"] is not None
67 ):
68 self.deterministic = meta_data["deterministic"]
69 if self.verifier is None and "verifier" in meta_data:
70 if isinstance(meta_data["verifier"], tuple): # File verifier
71 self.verifier = verifiers.mapping[meta_data["verifier"][0]](
72 Path(meta_data["verifier"][1])
73 )
74 elif meta_data["verifier"] in verifiers.mapping:
75 self.verifier = verifiers.mapping[meta_data["verifier"]]
76 if self.deterministic is None: # Default to False
77 self.deterministic = False
79 def __str__(self: Solver) -> str:
80 """Return the string representation of the solver."""
81 return self.name
83 def __repr__(self: Solver) -> str:
84 """Return detailed representation of the solver."""
85 return (
86 f"{self.name}:\n"
87 f"\t- Directory: {self.directory}\n"
88 f"\t- Deterministic: {self.deterministic}\n"
89 f"\t- Verifier: {self.verifier}\n"
90 f"\t- PCS File: {self.pcs_file}\n"
91 f"\t- Wrapper: {self.wrapper}"
92 )
94 def __eq__(self: Solver, other: Any) -> bool:
95 """Checks whether two solvers are equal."""
96 if isinstance(other, Solver):
97 return other.directory == self.directory
98 elif isinstance(other, str):
99 return other == self.name or Path(other) == self.directory
100 elif isinstance(other, Path):
101 return other == self.directory
102 return False
104 def __hash__(self: Solver) -> int:
105 """Pass to parent class hash function. Should be inherited but does not work without this."""
106 return super().__hash__()
108 @property
109 def pcs_file(self: Solver) -> Path:
110 """Get path of the parameter file."""
111 if self._pcs_file is None:
112 files = sorted(
113 [p for p in self.directory.iterdir() if p.is_file()],
114 key=lambda x: len(x.stem),
115 )
116 for file in files: # Loop through the files in ascending order of name (stem) length, to avoid selecting a Sparkle generated file
117 if file.name == Solver.meta_data:
118 continue # Skip this file, never correct
119 convention = PCSConverter.get_convention(file)
120 if convention != PCSConvention.UNKNOWN:
121 self._pcs_file = file
122 return self._pcs_file
123 return self._pcs_file
125 def get_pcs_file_type(self: Solver, convention: PCSConvention) -> Path:
126 """Get path of the parameter file of a specific convention."""
127 for file in self.directory.iterdir():
128 if file.name == Solver.meta_data:
129 continue # Skip this file, never correct
130 if PCSConverter.get_convention(file) == convention:
131 return file
132 return None
134 @property
135 def wrapper_extension(self: Solver) -> str:
136 """Get the extension of the wrapper file."""
137 if self._wrapper_extension is None:
138 # Determine which file is the wrapper by sorting alphabetically
139 wrapper = sorted(
140 [p for p in self.directory.iterdir() if p.stem == Solver._wrapper_file]
141 )[0]
142 self._wrapper_extension = wrapper.suffix
143 return self._wrapper_extension
145 @property
146 def wrapper(self: Solver) -> str:
147 """Get name of the wrapper file."""
148 return f"{Solver._wrapper_file}{self.wrapper_extension}"
150 @property
151 def wrapper_file(self: Solver) -> Path:
152 """Get path of the wrapper file."""
153 return self.directory / self.wrapper
155 def get_pcs_file(self: Solver, port_type: PCSConvention) -> Path:
156 """Get path of the parameter file of a specific convention.
158 Args:
159 port_type: Port type of the parameter file. If None, will return the
160 file with the shortest name.
162 Returns:
163 Path to the parameter file. None if it can not be resolved.
164 """
165 pcs_files = sorted([p for p in self.directory.iterdir() if p.suffix == ".pcs"])
166 if port_type is None:
167 return pcs_files[0]
168 for file in pcs_files:
169 if port_type == PCSConverter.get_convention(file):
170 return file
171 return None
173 def read_pcs_file(self: Solver) -> bool:
174 """Checks if the pcs file can be read."""
175 # TODO: Should be a .validate method instead
176 return PCSConverter.get_convention(self.pcs_file) is not None
178 def get_configuration_space(self: Solver) -> ConfigurationSpace:
179 """Get the ConfigurationSpace of the PCS file."""
180 if not self.pcs_file:
181 return None
182 return PCSConverter.parse(self.pcs_file)
184 def port_pcs(self: Solver, port_type: PCSConvention) -> None:
185 """Port the parameter file to the given port type."""
186 target_pcs_file = (
187 self.pcs_file.parent / f"{self.pcs_file.stem}_{port_type.name}.pcs"
188 )
189 if target_pcs_file.exists(): # Already exists, possibly user defined
190 return
191 PCSConverter.export(self.get_configuration_space(), port_type, target_pcs_file)
193 def build_cmd(
194 self: Solver,
195 instance: str | list[str],
196 objectives: list[SparkleObjective],
197 seed: int,
198 cutoff_time: int = None,
199 configuration: dict = None,
200 log_dir: Path = None,
201 ) -> list[str]:
202 """Build the solver call on an instance with a configuration.
204 Args:
205 instance: Path to the instance.
206 objectives: List of sparkle objectives.
207 seed: Seed of the solver.
208 cutoff_time: Cutoff time for the solver.
209 configuration: Configuration of the solver.
210 log_dir: Directory path for logs.
212 Returns:
213 List of commands and arguments to execute the solver.
214 """
215 if configuration is None:
216 configuration = {}
217 # Ensure configuration contains required entries for each wrapper
218 configuration["solver_dir"] = str(self.directory.absolute())
219 configuration["instance"] = instance
220 configuration["seed"] = seed
221 configuration["objectives"] = ",".join([str(obj) for obj in objectives])
222 configuration["cutoff_time"] = (
223 cutoff_time if cutoff_time is not None else sys.maxsize
224 )
225 if "configuration_id" in configuration:
226 del configuration["configuration_id"]
227 # Ensure stringification of dictionary will go correctly for key value pairs
228 configuration = {key: str(configuration[key]) for key in configuration}
229 solver_cmd = [
230 str(self.directory / self.wrapper),
231 f"'{json.dumps(configuration)}'",
232 ]
233 if log_dir is None:
234 log_dir = Path()
235 if cutoff_time is not None: # Use RunSolver
236 log_path_str = instance[0] if isinstance(instance, list) else instance
237 log_name_base = f"{Path(log_path_str).name}_{self.name}"
238 return RunSolver.wrap_command(
239 self.runsolver_exec,
240 solver_cmd,
241 cutoff_time,
242 log_dir,
243 log_name_base=log_name_base,
244 )
245 return solver_cmd
247 def run(
248 self: Solver,
249 instances: str | list[str] | InstanceSet | list[InstanceSet],
250 objectives: list[SparkleObjective],
251 seed: int,
252 cutoff_time: int = None,
253 configuration: dict = None,
254 run_on: Runner = Runner.LOCAL,
255 sbatch_options: list[str] = None,
256 slurm_prepend: str | list[str] | Path = None,
257 log_dir: Path = None,
258 ) -> SlurmRun | list[dict[str, Any]] | dict[str, Any]:
259 """Run the solver on an instance with a certain configuration.
261 Args:
262 instances: The instance(s) to run the solver on, list in case of multi-file.
263 In case of an instance set, will run on all instances in the set.
264 objectives: List of sparkle objectives.
265 seed: Seed to run the solver with. Fill with abitrary int in case of
266 determnistic solver.
267 cutoff_time: The cutoff time for the solver, measured through RunSolver.
268 If None, will be executed without RunSolver.
269 configuration: The solver configuration to use. Can be empty.
270 run_on: Whether to run on slurm or locally.
271 sbatch_options: The sbatch options to use.
272 slurm_prepend: The script to prepend to a slurm script.
273 log_dir: The log directory to use.
275 Returns:
276 Solver output dict possibly with runsolver values.
277 """
278 cmds = []
279 set_label = instances.name if isinstance(instances, InstanceSet) else "instances"
280 instances = [instances] if not isinstance(instances, list) else instances
281 log_dir = Path() if log_dir is None else log_dir
283 for instance in instances:
284 paths = (
285 instance.instance_paths
286 if isinstance(instance, InstanceSet)
287 else [instance]
288 )
289 for instance_path in paths:
290 instance_path = (
291 [str(p) for p in instance_path]
292 if isinstance(instance_path, list)
293 else instance_path
294 )
295 solver_cmd = self.build_cmd(
296 instance_path,
297 objectives=objectives,
298 seed=seed,
299 cutoff_time=cutoff_time,
300 configuration=configuration,
301 log_dir=log_dir,
302 )
303 cmds.append(" ".join(solver_cmd))
305 commandname = f"Run Solver {self.name} on {set_label}"
306 run = rrr.add_to_queue(
307 runner=run_on,
308 cmd=cmds,
309 name=commandname,
310 base_dir=log_dir,
311 sbatch_options=sbatch_options,
312 prepend=slurm_prepend,
313 )
315 if isinstance(run, LocalRun):
316 run.wait()
317 if run.status == Status.ERROR: # Subprocess resulted in error
318 print(f"WARNING: Solver {self.name} execution seems to have failed!\n")
319 for i, job in enumerate(run.jobs):
320 print(
321 f"[Job {i}] The used command was: {cmds[i]}\n"
322 "The error yielded was:\n"
323 f"\t-stdout: '{job.stdout}'\n"
324 f"\t-stderr: '{job.stderr}'\n"
325 )
326 return {
327 "status": SolverStatus.ERROR,
328 }
330 solver_outputs = []
331 for i, job in enumerate(run.jobs):
332 solver_cmd = cmds[i].split(" ")
333 solver_output = Solver.parse_solver_output(
334 run.jobs[i].stdout,
335 solver_call=solver_cmd,
336 objectives=objectives,
337 verifier=self.verifier,
338 )
339 solver_outputs.append(solver_output)
340 return solver_outputs if len(solver_outputs) > 1 else solver_output
341 return run
343 def run_performance_dataframe(
344 self: Solver,
345 instances: str | list[str] | InstanceSet,
346 performance_dataframe: PerformanceDataFrame,
347 config_ids: str | list[str] = None,
348 run_ids: list[int] | list[list[int]] = None,
349 cutoff_time: int = None,
350 objective: SparkleObjective = None,
351 train_set: InstanceSet = None,
352 sbatch_options: list[str] = None,
353 slurm_prepend: str | list[str] | Path = None,
354 dependencies: list[SlurmRun] = None,
355 log_dir: Path = None,
356 base_dir: Path = None,
357 job_name: str = None,
358 run_on: Runner = Runner.SLURM,
359 ) -> Run:
360 """Run the solver from and place the results in the performance dataframe.
362 This in practice actually runs Solver.run, but has a little script before/after,
363 to read and write to the performance dataframe.
365 Args:
366 instances: The instance(s) to run the solver on. In case of an instance set,
367 or list, will create a job for all instances in the set/list.
368 config_ids: The config indices to use in the performance dataframe.
369 performance_dataframe: The performance dataframe to use.
370 run_ids: List of run ids to use. If list of list, a list of runs is given
371 per instance. Otherwise, all runs are used for each instance.
372 cutoff_time: The cutoff time for the solver, measured through RunSolver.
373 objective: The objective to use, only relevant when determining the best
374 configuration.
375 train_set: The training set to use. If present, will determine the best
376 configuration of the solver using these instances and run with it on
377 all instances in the instance argument.
378 sbatch_options: List of slurm batch options to use
379 slurm_prepend: Slurm script to prepend to the sbatch
380 dependencies: List of slurm runs to use as dependencies
381 log_dir: Path where to place output files. Defaults to CWD.
382 base_dir: Path where to place output files.
383 job_name: Name of the job
384 If None, will generate a name based on Solver and Instances
385 run_on: On which platform to run the jobs. Default: Slurm.
387 Returns:
388 SlurmRun or Local run of the job.
389 """
390 instances = [instances] if isinstance(instances, str) else instances
391 set_name = "instances"
392 if isinstance(instances, InstanceSet):
393 set_name = instances.name
394 instances = [str(i) for i in instances.instance_paths]
395 if not isinstance(config_ids, list):
396 config_ids = [config_ids]
397 configurations = [
398 performance_dataframe.get_full_configuration(str(self.directory), config_id)
399 if config_id
400 else None
401 for config_id in config_ids
402 ]
403 if run_ids is None:
404 run_ids = performance_dataframe.run_ids
405 if isinstance(run_ids[0], list): # Runs per instance
406 combinations = []
407 for index, instance in enumerate(instances):
408 for run_id in run_ids[index]:
409 combinations.extend(
410 [
411 (instance, config_id, config, run_id)
412 for config_id, config in zip(config_ids, configurations)
413 ]
414 )
415 else: # Runs for all instances
416 import itertools
418 combinations = [
419 (instance, config_data[0], config_data[1], run_id)
420 for instance, config_data, run_id in itertools.product(
421 instances,
422 zip(config_ids, configurations),
423 performance_dataframe.run_ids,
424 )
425 ]
426 objective_arg = f"--target-objective {objective.name}" if objective else ""
427 train_arg = (
428 "--best-configuration-instances "
429 + " ".join(
430 f"{set_name},{instance_name}"
431 for set_name, instance_name in train_set.instance_pairs
432 )
433 if train_set
434 else ""
435 )
436 configuration_args = [
437 ""
438 if not config_id and not config
439 else f"--configuration-id {config_id}"
440 if not config
441 else f"--configuration '{json.dumps(config)}'"
442 for _, config_id, config, _ in combinations
443 ]
445 # We run all instances/configs/runs combinations
446 # For each value we try to resolve from the PDF, to avoid high read loads during executions
447 cmds = [
448 f"python3 {Solver.solver_cli} "
449 f"--solver {self.directory} "
450 f"--instance {instance} "
451 f"{config_arg} "
452 # f"{'--configuration-id ' + config_id if not config else '--configuration"' + str(config) + '\"'} "
453 f"--run-index {run_id} "
454 f"--objectives {' '.join([obj.name for obj in performance_dataframe.objectives])} "
455 f"--performance-dataframe {performance_dataframe.csv_filepath} "
456 f"--cutoff-time {cutoff_time} "
457 f"--log-dir {log_dir} "
458 f"--seed {random.randint(0, 2**32 - 1)} "
459 f"{objective_arg} "
460 f"{train_arg}"
461 for (instance, _, _, run_id), config_arg in zip(
462 combinations, configuration_args
463 )
464 ]
465 job_name = (
466 f"Run Solver {self.name} on {set_name}" if job_name is None else job_name
467 )
468 r = rrr.add_to_queue(
469 runner=run_on,
470 cmd=cmds,
471 name=job_name,
472 base_dir=base_dir,
473 sbatch_options=sbatch_options,
474 prepend=slurm_prepend,
475 dependencies=dependencies,
476 )
477 if run_on == Runner.LOCAL:
478 r.wait()
479 return r
481 @staticmethod
482 def config_str_to_dict(config_str: str) -> dict[str, str]:
483 """Parse a configuration string to a dictionary."""
484 # First we filter the configuration of unwanted characters
485 config_str = config_str.strip().replace("-", "")
486 # Then we split the string by spaces, but conserve substrings
487 config_list = shlex.split(config_str)
488 # We return empty for empty input OR uneven input
489 if config_str == "" or config_str == r"{}" or len(config_list) & 1:
490 return {}
491 config_dict = {}
492 for index in range(0, len(config_list), 2):
493 # As the value will already be a string object, no quotes are allowed in it
494 value = config_list[index + 1].strip('"').strip("'")
495 config_dict[config_list[index]] = value
496 return config_dict
498 @staticmethod
499 def parse_solver_output(
500 solver_output: str,
501 solver_call: list[str | Path] = None,
502 objectives: list[SparkleObjective] = None,
503 verifier: verifiers.SolutionVerifier = None,
504 ) -> dict[str, Any]:
505 """Parse the output of the solver.
507 Args:
508 solver_output: The output of the solver run which needs to be parsed
509 solver_call: The solver call used to run the solver
510 objectives: The objectives to apply to the solver output
511 verifier: The verifier to check the solver output
513 Returns:
514 Dictionary representing the parsed solver output
515 """
516 used_runsolver = False
517 if (
518 solver_call is not None
519 and len(solver_call) > 2
520 and solver_call[0].endswith("runsolver")
521 or solver_call[1].endswith("py_runsolver.py")
522 ):
523 used_runsolver = True # PyRunsolver or RunSolver was used
524 parsed_output = RunSolver.get_solver_output(solver_call, solver_output)
525 else:
526 parsed_output = ast.literal_eval(solver_output)
527 # cast status attribute from str to Enum
528 parsed_output["status"] = SolverStatus(parsed_output["status"])
529 # Apply objectives to parsed output, runtime based objectives added here
530 if verifier is not None and used_runsolver:
531 # Horrible hack to get the instance from the solver input
532 solver_call_str: str = " ".join(solver_call)
533 solver_input_str = solver_call_str.split(Solver._wrapper_file, maxsplit=1)[1]
534 solver_input_str = solver_input_str.split(" ", maxsplit=1)[1]
535 solver_input_str = solver_input_str[
536 solver_input_str.index("{") : solver_input_str.index("}") + 1
537 ]
538 solver_input = ast.literal_eval(solver_input_str)
539 target_instance = Path(solver_input["instance"])
540 parsed_output["status"] = verifier.verify(
541 target_instance, parsed_output, solver_call
542 )
544 # Create objective map
545 objectives = {o.stem: o for o in objectives} if objectives else {}
546 removable_keys = ["cutoff_time"] # Keys to remove
548 # apply objectives to parsed output, runtime based objectives added here
549 for key, value in parsed_output.items():
550 if objectives and key in objectives:
551 objective = objectives[key]
552 removable_keys.append(key) # We translate it into the full name
553 else:
554 objective = resolve_objective(key)
555 # If not found in objectives, resolve to which objective the output belongs
556 if objective is None: # Could not parse, skip
557 continue
558 if objective.use_time == UseTime.NO:
559 if objective.post_process is not None:
560 parsed_output[key] = objective.post_process(value)
561 else:
562 if not used_runsolver:
563 continue
564 if objective.use_time == UseTime.CPU_TIME:
565 parsed_output[key] = parsed_output["cpu_time"]
566 else:
567 parsed_output[key] = parsed_output["wall_time"]
568 if objective.post_process is not None:
569 parsed_output[key] = objective.post_process(
570 parsed_output[key],
571 parsed_output["cutoff_time"],
572 parsed_output["status"],
573 )
575 # Replace or remove keys based on the objective names
576 for key in removable_keys:
577 if key in parsed_output:
578 if key in objectives:
579 # Map the result to the objective
580 parsed_output[objectives[key].name] = parsed_output[key]
581 if key != objectives[key].name: # Only delete actual mappings
582 del parsed_output[key]
583 else:
584 del parsed_output[key]
585 return parsed_output