Source code for sparkle.platform.settings_objects

"""Classes and Enums to control settings."""

from __future__ import annotations
import configparser
import argparse
from enum import Enum
from pathlib import Path
from typing import Any, NamedTuple, Optional

from runrunner import Runner

from sparkle.types import SparkleObjective, resolve_objective
from sparkle.configurator.configurator import Configurator
from sparkle.configurator import implementations as cim
from sparkle.platform.cli_types import VerbosityLevel


[docs] class Option(NamedTuple): """Class to define an option in the Settings.""" name: str section: str type: Any default_value: Any alternatives: tuple[str, ...] help: str = "" cli_kwargs: dict[str, Any] = {} def __str__(self: Option) -> str: """Return the option name.""" return self.name def __eq__(self: Option, other: Any) -> bool: """Check if two options are equal.""" if isinstance(other, Option): return ( self.name == other.name and self.section == other.section and self.type == other.type and self.default_value == other.default_value and self.alternatives == other.alternatives ) if isinstance(other, str): return self.name == other or other in self.alternatives return False @property def args(self: Option) -> list[str]: """Return the option names as a command line arguments.""" return [ f"--{name.replace('_', '-')}" for name in [self.name] + list(self.alternatives) ] @property def kwargs(self: Option) -> dict[str, Any]: """Return the option attributes as kwargs.""" kw = {"help": self.help, **self.cli_kwargs} # If this option uses a boolean flag action, argparse must NOT receive 'type' action = kw.get("action") if action in ("store_true", "store_false"): return kw # Otherwise include the base 'type' return {"type": self.type, **kw}
[docs] class Settings: """Class to read, write, set, and get settings.""" # CWD Prefix cwd_prefix = Path() # Empty for now # Library prefix lib_prefix = Path(__file__).parent.parent.resolve() # Default directory names rawdata_dir = Path("Raw_Data") analysis_dir = Path("Analysis") DEFAULT_settings_dir = Path("Settings") __settings_file = Path("sparkle_settings.ini") __latest_settings_file = Path("latest.ini") # Default settings path DEFAULT_settings_path = Path(cwd_prefix / DEFAULT_settings_dir / __settings_file) DEFAULT_previous_settings_path = Path( cwd_prefix / DEFAULT_settings_dir / __latest_settings_file ) DEFAULT_reference_dir = DEFAULT_settings_dir / "Reference_Lists" # Default library pathing DEFAULT_components = lib_prefix / "Components" # Report Component: Bilbiography bibliography_path = DEFAULT_components / "latex_source" / "report.bib" # Example settings path DEFAULT_example_settings_path = Path(DEFAULT_components / "sparkle_settings.ini") # Runsolver component DEFAULT_runsolver_dir = DEFAULT_components / "runsolver" / "src" DEFAULT_runsolver_exec = DEFAULT_runsolver_dir / "runsolver" # Ablation component DEFAULT_ablation_dir = DEFAULT_components / "ablationAnalysis-0.9.4" DEFAULT_ablation_exec = DEFAULT_ablation_dir / "ablationAnalysis" DEFAULT_ablation_validation_exec = DEFAULT_ablation_dir / "ablationValidation" # Default input directory pathing DEFAULT_solver_dir = cwd_prefix / "Solvers" DEFAULT_instance_dir = cwd_prefix / "Instances" DEFAULT_extractor_dir = cwd_prefix / "Extractors" DEFAULT_snapshot_dir = cwd_prefix / "Snapshots" # Default output directory pathing DEFAULT_tmp_output = cwd_prefix / "Tmp" DEFAULT_output = cwd_prefix / "Output" DEFAULT_configuration_output = DEFAULT_output / "Configuration" DEFAULT_selection_output = DEFAULT_output / "Selection" DEFAULT_parallel_portfolio_output = DEFAULT_output / "Parallel_Portfolio" DEFAULT_ablation_output = DEFAULT_output / "Ablation" DEFAULT_log_output = DEFAULT_output / "Log" # Default output subdirs DEFAULT_output_analysis = DEFAULT_output / analysis_dir # Old default output dirs which should be part of something else DEFAULT_feature_data = DEFAULT_output / "Feature_Data" DEFAULT_performance_data = DEFAULT_output / "Performance_Data" # Collection of all working dirs for platform DEFAULT_working_dirs = [ DEFAULT_solver_dir, DEFAULT_instance_dir, DEFAULT_extractor_dir, DEFAULT_output, DEFAULT_configuration_output, DEFAULT_selection_output, DEFAULT_output_analysis, DEFAULT_tmp_output, DEFAULT_log_output, DEFAULT_feature_data, DEFAULT_performance_data, DEFAULT_settings_dir, DEFAULT_reference_dir, ] # Old default file paths from GV which should be turned into variables DEFAULT_feature_data_path = DEFAULT_feature_data / "feature_data.csv" DEFAULT_performance_data_path = DEFAULT_performance_data / "performance_data.csv" # Define sections and options # GENERAL Options SECTION_general: str = "general" OPTION_objectives = Option( "objectives", SECTION_general, str, None, ("sparkle_objectives",), "A list of Sparkle objectives.", cli_kwargs={"nargs": "+"}, ) OPTION_configurator = Option( "configurator", SECTION_general, str, None, tuple(), "Name of the configurator to use.", ) OPTION_solver_cutoff_time = Option( "solver_cutoff_time", SECTION_general, int, None, ("target_cutoff_time", "cutoff_time_each_solver_call"), "Solver cutoff time in seconds.", ) OPTION_extractor_cutoff_time = Option( "extractor_cutoff_time", SECTION_general, int, None, tuple("cutoff_time_each_feature_computation"), "Extractor cutoff time in seconds.", ) OPTION_run_on = Option( "run_on", SECTION_general, Runner, None, tuple(), "On which compute resource to execute.", cli_kwargs={"choices": [Runner.LOCAL, Runner.SLURM]}, ) OPTION_verbosity = Option( "verbosity", SECTION_general, VerbosityLevel, VerbosityLevel.STANDARD, ("verbosity_level",), "Verbosity level.", ) OPTION_seed = Option( "seed", SECTION_general, int, None, tuple(), "Seed to use for pseudo-random number generators.", ) OPTION_appendices = Option( "appendices", SECTION_general, bool, False, tuple(), "Include the appendix section in the generated report.", cli_kwargs={ "action": "store_true", "default": None, }, ) # CONFIGURATION Options SECTION_configuration = "configuration" OPTION_configurator_number_of_runs = Option( "number_of_runs", SECTION_configuration, int, None, tuple(), "The number of independent configurator jobs/runs.", ) OPTION_configurator_solver_call_budget = Option( "solver_calls", SECTION_configuration, int, None, tuple(), "The maximum number of calls (evaluations) a configurator can do in a single " "run of the solver.", ) OPTION_configurator_max_iterations = Option( "max_iterations", SECTION_configuration, int, None, ("maximum_iterations",), "The maximum number of iterations a configurator can do in a single job.", ) # ABLATION Options SECTION_ablation = "ablation" OPTION_ablation_racing = Option( "racing", SECTION_ablation, bool, False, ("ablation_racing",), "Set a flag indicating whether racing should be used for ablation.", ) OPTION_ablation_clis_per_node = Option( "clis_per_node", SECTION_ablation, int, None, ( "max_parallel_runs_per_node", "maximum_parallel_runs_per_node", ), "The maximum number of ablation analysis jobs to run in parallel on a single " "node.", ) # SELECTION Options SECTION_selection = "selection" OPTION_selection_class = Option( "selector_class", SECTION_selection, str, None, ("class",), "Can contain any of the class names as defined in asf.selectors.", ) OPTION_selection_model = Option( "selector_model", SECTION_selection, str, None, ("model",), "Can be any of the sklearn.ensemble models.", ) OPTION_minimum_marginal_contribution = Option( "minimum_marginal_contribution", SECTION_selection, float, 0.01, ( "minimum_marginal_contribution", "minimum_contribution", "contribution_threshold", ), "The minimum marginal contribution a solver (configuration) must have to be used for the selector.", ) # SMAC2 Options SECTION_smac2 = "smac2" OPTION_smac2_wallclock_time_budget = Option( "wallclock_time_budget", SECTION_smac2, int, None, ("wallclock_time",), "The wallclock time budget in seconds for each SMAC2 run.", ) OPTION_smac2_cpu_time_budget = Option( "cpu_time_budget", SECTION_smac2, int, None, ("cpu_time",), "The cpu time budget in seconds for each SMAC2 run.", ) OPTION_smac2_target_cutoff_length = Option( "target_cutoff_length", SECTION_smac2, str, None, ("cutoff_length", "solver_cutoff_length"), "The target cutoff length for SMAC2 solver call.", ) OPTION_smac2_count_tuner_time = Option( "use_cpu_time_in_tunertime", SECTION_smac2, bool, None, ("countSMACTimeAsTunerTime",), "Whether to count and deducted SMAC2 CPU time from the CPU time budget.", ) OPTION_smac2_cli_cores = Option( "cli_cores", SECTION_smac2, int, None, tuple(), "Number of cores to use to execute SMAC2 runs.", ) OPTION_smac2_max_iterations = Option( "max_iterations", SECTION_smac2, int, None, ( "iteration_limit", "numIterations", "numberOfIterations", ), "The maximum number of iterations SMAC2 configurator can do in a single job.", ) # SMAC3 Options SECTION_smac3 = "smac3" OPTION_smac3_number_of_trials = Option( "n_trials", SECTION_smac3, int, None, ("n_trials", "number_of_trials", "solver_calls"), "Maximum calls SMAC3 is allowed to make to the Solver in a single run/job.", ) OPTION_smac3_facade = Option( "facade", SECTION_smac3, str, "AlgorithmConfigurationFacade", ("facade", "smac_facade", "smac3_facade"), "The SMAC3 Facade to use. See the SMAC3 documentation for more options.", ) OPTION_smac3_facade_max_ratio = Option( "facade_max_ratio", SECTION_smac3, float, None, ("facade_max_ratio", "smac3_facade_max_ratio", "smac3_facade_max_ratio"), "The SMAC3 Facade max ratio. See the SMAC3 documentation for more options.", ) OPTION_smac3_crash_cost = Option( "crash_cost", SECTION_smac3, float, None, tuple(), "Defines the cost for a failed trial, defaults in SMAC3 to np.inf.", ) OPTION_smac3_termination_cost_threshold = Option( "termination_cost_threshold", SECTION_smac3, float, None, tuple(), "Defines a cost threshold when the SMAC3 optimization should stop.", ) OPTION_smac3_wallclock_time_budget = Option( "walltime_limit", SECTION_smac3, float, None, ("wallclock_time", "wallclock_budget", "wallclock_time_budget"), "The maximum time in seconds that SMAC3 is allowed to run per job.", ) OPTION_smac3_cpu_time_budget = Option( "cputime_limit", SECTION_smac3, float, None, ("cpu_time", "cpu_budget", "cpu_time_budget"), "The maximum CPU time in seconds that SMAC3 is allowed to run per job.", ) OPTION_smac3_use_default_config = Option( "use_default_config", SECTION_smac3, bool, None, tuple(), "If True, the configspace's default configuration is evaluated in the initial " "design. For historic benchmark reasons, this is False by default. Notice, that " "this will result in n_configs + 1 for the initial design. Respecting n_trials, " "this will result in one fewer evaluated configuration in the optimization.", ) OPTION_smac3_min_budget = Option( "min_budget", SECTION_smac3, float, None, ("minimum_budget",), "The minimum budget (epochs, subset size, number of instances, ...) that is used" " for the optimization. Use this argument if you use multi-fidelity or instance " "optimization.", ) OPTION_smac3_max_budget = Option( "max_budget", SECTION_smac3, float, None, ("maximum_budget",), "The maximum budget (epochs, subset size, number of instances, ...) that is used" " for the optimization. Use this argument if you use multi-fidelity or instance " "optimization.", ) # IRACE Options SECTION_irace = "irace" OPTION_irace_max_time = Option( "max_time", SECTION_irace, int, 0, ("maximum_time",), "The maximum time in seconds for each IRACE run/job.", ) OPTION_irace_max_experiments = Option( "max_experiments", SECTION_irace, int, 0, ("maximum_experiments",), "The maximum number of experiments for each IRACE run/job.", ) OPTION_irace_first_test = Option( "first_test", SECTION_irace, int, None, tuple(), "Specifies how many instances are evaluated before the first elimination test. " "IRACE Default: 5.", ) OPTION_irace_mu = Option( "mu", SECTION_irace, int, None, tuple(), "Parameter used to define the number of configurations sampled and evaluated at " "each iteration. IRACE Default: 5.", ) OPTION_irace_max_iterations = Option( "max_iterations", SECTION_irace, int, None, ("nb_iterations", "iterations", "max_iterations"), "Maximum number of iterations to be executed. Each iteration involves the " "generation of new configurations and the use of racing to select the best " "configurations. By default (with 0), irace calculates a minimum number of " "iterations as N^iter = ⌊2 + log2 N param⌋, where N^param is the number of " "non-fixed parameters to be tuned. Setting this parameter may make irace stop " "sooner than it should without using all the available budget. IRACE recommends" " to use the default value (Empty).", ) # ParamILS Options SECTION_paramils = "paramils" OPTION_paramils_min_runs = Option( "min_runs", SECTION_paramils, int, None, ("minimum_runs",), "Set the minimum number of runs for ParamILS for each run/job.", ) OPTION_paramils_max_runs = Option( "max_runs", SECTION_paramils, int, None, ("maximum_runs",), "Set the maximum number of runs for ParamILS for each run/job.", ) OPTION_paramils_cpu_time_budget = Option( "cputime_budget", SECTION_paramils, int, None, ( "cputime_limit", "cputime_limit", "tunertime_limit", "tuner_timeout", "tunerTimeout", ), "The maximum CPU time for each ParamILS run/job.", ) OPTION_paramils_random_restart = Option( "random_restart", SECTION_paramils, float, None, tuple(), "Set the random restart chance for ParamILS.", ) OPTION_paramils_focused = Option( "focused_approach", SECTION_paramils, bool, False, ("focused",), "Set the focused approach for ParamILS.", ) OPTION_paramils_count_tuner_time = Option( "use_cpu_time_in_tunertime", SECTION_paramils, bool, None, tuple(), "Whether to count and deducted ParamILS CPU time from the CPU time budget.", ) OPTION_paramils_cli_cores = Option( "cli_cores", SECTION_paramils, int, None, tuple(), "Number of cores to use for ParamILS runs.", ) OPTION_paramils_max_iterations = Option( "max_iterations", SECTION_paramils, int, None, ( "iteration_limit", "numIterations", "numberOfIterations", "maximum_iterations", ), "The maximum number of ParamILS iterations per run/job.", ) OPTION_paramils_number_initial_configurations = Option( "initial_configurations", SECTION_paramils, int, None, "The number of initial configurations ParamILS should evaluate.", ) SECTION_parallel_portfolio = "parallel_portfolio" OPTION_parallel_portfolio_check_interval = Option( "check_interval", SECTION_parallel_portfolio, int, None, tuple(), "The interval time in seconds when Solvers are checked for their status.", ) OPTION_parallel_portfolio_number_of_seeds_per_solver = Option( "num_seeds_per_solver", SECTION_parallel_portfolio, int, None, ("solver_seeds",), "The number of seeds per solver.", ) SECTION_slurm = "slurm" OPTION_slurm_parallel_jobs = Option( "number_of_jobs_in_parallel", SECTION_slurm, int, None, ("num_job_in_parallel",), "The number of jobs to run in parallel.", ) OPTION_slurm_prepend_script = Option( "prepend_script", SECTION_slurm, str, None, ("job_prepend", "prepend"), "Slurm script to prepend to the sbatch.", ) sections_options: dict[str, list[Option]] = { SECTION_general: [ OPTION_objectives, OPTION_configurator, OPTION_solver_cutoff_time, OPTION_extractor_cutoff_time, OPTION_run_on, OPTION_appendices, OPTION_verbosity, OPTION_seed, ], SECTION_configuration: [ OPTION_configurator_number_of_runs, OPTION_configurator_solver_call_budget, OPTION_configurator_max_iterations, ], SECTION_ablation: [ OPTION_ablation_racing, OPTION_ablation_clis_per_node, ], SECTION_selection: [ OPTION_selection_class, OPTION_selection_model, OPTION_minimum_marginal_contribution, ], SECTION_smac2: [ OPTION_smac2_wallclock_time_budget, OPTION_smac2_cpu_time_budget, OPTION_smac2_target_cutoff_length, OPTION_smac2_count_tuner_time, OPTION_smac2_cli_cores, OPTION_smac2_max_iterations, ], SECTION_smac3: [ OPTION_smac3_number_of_trials, OPTION_smac3_facade, OPTION_smac3_facade_max_ratio, OPTION_smac3_crash_cost, OPTION_smac3_termination_cost_threshold, OPTION_smac3_wallclock_time_budget, OPTION_smac3_cpu_time_budget, OPTION_smac3_use_default_config, OPTION_smac3_min_budget, OPTION_smac3_max_budget, ], SECTION_irace: [ OPTION_irace_max_time, OPTION_irace_max_experiments, OPTION_irace_first_test, OPTION_irace_mu, OPTION_irace_max_iterations, ], SECTION_paramils: [ OPTION_paramils_min_runs, OPTION_paramils_max_runs, OPTION_paramils_cpu_time_budget, OPTION_paramils_random_restart, OPTION_paramils_focused, OPTION_paramils_count_tuner_time, OPTION_paramils_cli_cores, OPTION_paramils_max_iterations, OPTION_paramils_number_initial_configurations, ], SECTION_parallel_portfolio: [ OPTION_parallel_portfolio_check_interval, OPTION_parallel_portfolio_number_of_seeds_per_solver, ], SECTION_slurm: [OPTION_slurm_parallel_jobs, OPTION_slurm_prepend_script], } def __init__( self: Settings, file_path: Path, argsv: argparse.Namespace = None ) -> None: """Initialise a settings object. Args: file_path (Path): Path to the settings file. argsv: The CLI arguments to process. """ # Settings 'dictionary' in configparser format self.__settings = configparser.ConfigParser() for section in self.sections_options.keys(): self.__settings.add_section(section) self.__settings[section] = {} # General attributes self.__sparkle_objectives: list[SparkleObjective] = None self.__general_sparkle_configurator: Configurator = None self.__solver_cutoff_time: int = None self.__extractor_cutoff_time: int = None self.__run_on: Runner = None self.__appendices: bool = False self.__verbosity_level: VerbosityLevel = None self.__seed: Optional[int] = None # Configuration attributes self.__configurator_solver_call_budget: int = None self.__configurator_number_of_runs: int = None self.__configurator_max_iterations: int = None # Ablation attributes self.__ablation_racing_flag: bool = None self.__ablation_max_parallel_runs_per_node: int = None # Selection attributes self.__selection_model: str = None self.__selection_class: str = None self.__minimum_marginal_contribution: float = None # SMAC2 attributes self.__smac2_wallclock_time_budget: int = None self.__smac2_cpu_time_budget: int = None self.__smac2_target_cutoff_length: str = None self.__smac2_use_tunertime_in_cpu_time_budget: bool = None self.__smac2_cli_cores: int = None self.__smac2_max_iterations: int = None # SMAC3 attributes self.__smac3_number_of_trials: int = None self.__smac3_facade: str = None self.__smac3_facade_max_ratio: float = None self.__smac3_crash_cost: float = None self.__smac3_termination_cost_threshold: float = None self.__smac3_wallclock_time_limit: int = None self.__smac3_cputime_limit: int = None self.__smac3_use_default_config: bool = None self.__smac3_min_budget: float = None self.__smac3_max_budget: float = None # IRACE attributes self.__irace_max_time: int = None self.__irace_max_experiments: int = None self.__irace_first_test: int = None self.__irace_mu: int = None self.__irace_max_iterations: int = None # ParamILS attributes self.__paramils_cpu_time_budget: int = None self.__paramils_min_runs: int = None self.__paramils_max_runs: int = None self.__paramils_random_restart: float = None self.__paramils_focused_approach: bool = None self.__paramils_use_cpu_time_in_tunertime: bool = None self.__paramils_cli_cores: int = None self.__paramils_max_iterations: int = None self.__paramils_number_initial_configurations: int = None # Parallel portfolio attributes self.__parallel_portfolio_check_interval: int = None self.__parallel_portfolio_num_seeds_per_solver: int = None # Slurm attributes self.__slurm_jobs_in_parallel: int = None self.__slurm_job_prepend: str = None # The seed that has been used to set the random state self.random_state: Optional[int] = None if file_path and file_path.exists(): self.read_settings_ini(file_path) if argsv: self.apply_arguments(argsv)
[docs] def read_settings_ini(self: Settings, file_path: Path) -> None: """Read the settings from an INI file.""" if not file_path.exists(): raise ValueError(f"Settings file {file_path} does not exist.") # Read file file_settings = configparser.ConfigParser() file_settings.read(file_path) # Set internal settings based on data read from FILE if they were read # successfully if file_settings.sections() == []: # Print error if unable to read the settings print( f"ERROR: Failed to read settings from {file_path} The file may " "have been empty or be in another format than INI." ) return for section in file_settings.sections(): if section not in self.__settings.sections(): print(f'Unrecognised section: "{section}" in file {file_path} ignored') continue for option_name in file_settings.options(section): if option_name not in self.sections_options[section]: if section == Settings.SECTION_slurm: # Flexible section self.__settings.set( section, option_name, file_settings.get(section, option_name), ) else: print( f'Unrecognised section - option combination: "{section} ' f'{option_name}" in file {file_path} ignored' ) continue option_index = self.sections_options[section].index(option_name) option = self.sections_options[section][option_index] self.__settings.set( section, option.name, file_settings.get(section, option_name) ) del file_settings
[docs] def write_settings_ini(self: Settings, file_path: Path) -> None: """Write the settings to an INI file.""" # Create needed directories if they don't exist file_path.parent.mkdir(parents=True, exist_ok=True) # We don't write empty sections for s in self.__settings.sections(): if not self.__settings[s]: self.__settings.remove_section(s) with file_path.open("w") as fout: self.__settings.write(fout) for s in self.sections_options.keys(): if s not in self.__settings.sections(): self.__settings.add_section(s)
[docs] def write_used_settings(self: Settings) -> None: """Write the used settings to the default locations.""" # Write to latest settings file self.write_settings_ini(self.DEFAULT_previous_settings_path)
[docs] def apply_arguments(self: Settings, argsv: argparse.Namespace) -> None: """Apply the arguments to the settings.""" # Read a possible second file, that overwrites the first, where applicable # e.g. settings are not deleted, but overwritten where applicable if hasattr(argsv, "settings_file") and argsv.settings_file: self.read_settings_ini(argsv.settings_file) # Match all possible arguments to the settings for argument in argsv.__dict__.keys(): value = argsv.__dict__[argument] if value is None: continue # Skip None value = value.name if isinstance(value, Enum) else str(value) for section in self.sections_options.keys(): if argument in self.sections_options[section]: index = self.sections_options[section].index(argument) option = self.sections_options[section][index] self.__settings.set(option.section, option.name, value) break
def _abstract_getter(self: Settings, option: Option) -> Any: """Abstract getter method.""" if self.__settings.has_option(option.section, option.name): if option.type is bool: return self.__settings.getboolean(option.section, option.name) value = self.__settings.get(option.section, option.name) if not isinstance(value, option.type): if issubclass(option.type, Enum): return option.type[value.upper()] return option.type(value) # Attempt to resolve str to type return value return option.default_value # General settings ### @property def objectives(self: Settings) -> list[SparkleObjective]: """Get the objectives for Sparkle.""" if self.__sparkle_objectives is None and self.__settings.has_option( Settings.SECTION_general, "objectives" ): objectives = self.__settings[Settings.SECTION_general]["objectives"] if "status" not in objectives: objectives += ",status:metric" if "cpu_time" not in objectives: objectives += ",cpu_time:metric" if "wall_time" not in objectives: objectives += ",wall_time:metric" if "memory" not in objectives: objectives += ",memory:metric" self.__sparkle_objectives = [ resolve_objective(obj) for obj in objectives.split(",") ] return self.__sparkle_objectives @property def configurator(self: Settings) -> Configurator: """Get the configurator class (instance).""" if self.__general_sparkle_configurator is None and self.__settings.has_option( Settings.OPTION_configurator.section, Settings.OPTION_configurator.name ): self.__general_sparkle_configurator = cim.resolve_configurator( self.__settings.get( Settings.OPTION_configurator.section, Settings.OPTION_configurator.name, ) )() return self.__general_sparkle_configurator @property def solver_cutoff_time(self: Settings) -> int: """Solver cutoff time in seconds.""" if self.__solver_cutoff_time is None: self.__solver_cutoff_time = self._abstract_getter( Settings.OPTION_solver_cutoff_time ) return self.__solver_cutoff_time @property def extractor_cutoff_time(self: Settings) -> int: """Extractor cutoff time in seconds.""" if self.__extractor_cutoff_time is None: self.__extractor_cutoff_time = self._abstract_getter( Settings.OPTION_extractor_cutoff_time ) return self.__extractor_cutoff_time @property def run_on(self: Settings) -> Runner: """On which compute to run (Local or Slurm).""" if self.__run_on is None: self.__run_on = self._abstract_getter(Settings.OPTION_run_on) return self.__run_on @property def appendices(self: Settings) -> bool: """Whether to include appendices in the report.""" return self._abstract_getter(Settings.OPTION_appendices) @property def verbosity_level(self: Settings) -> VerbosityLevel: """Verbosity level to use in CLI commands.""" if self.__verbosity_level is None: if self.__settings.has_option( Settings.OPTION_verbosity.section, Settings.OPTION_verbosity.name ): self.__verbosity_level = VerbosityLevel[ self.__settings.get( Settings.OPTION_verbosity.section, Settings.OPTION_verbosity.name, ) ] else: self.__verbosity_level = Settings.OPTION_verbosity.default_value return self.__verbosity_level @property def seed(self: Settings) -> int: """Seed to use in CLI commands.""" if self.__seed is not None: return self.__seed section, name = Settings.OPTION_seed.section, Settings.OPTION_seed.name if self.__settings.has_option(section, name): value = self.__settings.get(section, name) self.__seed = int(value) else: self.__seed = Settings.OPTION_seed.default_value return self.__seed @seed.setter def seed(self: Settings, value: int) -> None: """Set the seed value (overwrites settings).""" self.__seed = value self.__settings.set( Settings.OPTION_seed.section, Settings.OPTION_seed.name, str(self.__seed) ) # Configuration settings ### @property def configurator_solver_call_budget(self: Settings) -> int: """The amount of calls a configurator can do to the solver.""" if self.__configurator_solver_call_budget is None: self.__configurator_solver_call_budget = self._abstract_getter( Settings.OPTION_configurator_solver_call_budget ) return self.__configurator_solver_call_budget @property def configurator_number_of_runs(self: Settings) -> int: """Get the amount of configurator runs to do.""" if self.__configurator_number_of_runs is None: self.__configurator_number_of_runs = self._abstract_getter( Settings.OPTION_configurator_number_of_runs ) return self.__configurator_number_of_runs @property def configurator_max_iterations(self: Settings) -> int: """Get the amount of configurator iterations to do.""" if self.__configurator_max_iterations is None: self.__configurator_max_iterations = self._abstract_getter( Settings.OPTION_configurator_max_iterations ) return self.__configurator_max_iterations # Ablation settings ### @property def ablation_racing_flag(self: Settings) -> bool: """Get the ablation racing flag.""" if self.__ablation_racing_flag is None: self.__ablation_racing_flag = self._abstract_getter( Settings.OPTION_ablation_racing ) return self.__ablation_racing_flag @property def ablation_max_parallel_runs_per_node(self: Settings) -> int: """Get the ablation max parallel runs per node.""" if self.__ablation_max_parallel_runs_per_node is None: self.__ablation_max_parallel_runs_per_node = self._abstract_getter( Settings.OPTION_ablation_clis_per_node ) return self.__ablation_max_parallel_runs_per_node # Selection settings ### @property def selection_model(self: Settings) -> str: """Get the selection model.""" if self.__selection_model is None: self.__selection_model = self._abstract_getter( Settings.OPTION_selection_model ) return self.__selection_model @property def selection_class(self: Settings) -> str: """Get the selection class.""" if self.__selection_class is None: self.__selection_class = self._abstract_getter( Settings.OPTION_selection_class ) return self.__selection_class @property def minimum_marginal_contribution(self: Settings) -> float: """Get the minimum marginal contribution.""" if self.__minimum_marginal_contribution is None: self.__minimum_marginal_contribution = self._abstract_getter( Settings.OPTION_minimum_marginal_contribution ) return self.__minimum_marginal_contribution # Configuration: SMAC2 specific settings ### @property def smac2_wallclock_time_budget(self: Settings) -> int: """Return the SMAC2 wallclock budget per configuration run in seconds.""" if self.__smac2_wallclock_time_budget is None: self.__smac2_wallclock_time_budget = self._abstract_getter( Settings.OPTION_smac2_wallclock_time_budget ) return self.__smac2_wallclock_time_budget @property def smac2_cpu_time_budget(self: Settings) -> int: """Return the SMAC2 CPU budget per configuration run in seconds.""" if self.__smac2_cpu_time_budget is None: self.__smac2_cpu_time_budget = self._abstract_getter( Settings.OPTION_smac2_cpu_time_budget ) return self.__smac2_cpu_time_budget @property def smac2_target_cutoff_length(self: Settings) -> str: """Return the SMAC2 target cutoff length.""" if self.__smac2_target_cutoff_length is None: self.__smac2_target_cutoff_length = self._abstract_getter( Settings.OPTION_smac2_target_cutoff_length ) return self.__smac2_target_cutoff_length @property def smac2_use_tunertime_in_cpu_time_budget(self: Settings) -> bool: """Return whether SMAC2 time should be used in CPU time budget.""" if self.__smac2_use_tunertime_in_cpu_time_budget is None: self.__smac2_use_tunertime_in_cpu_time_budget = self._abstract_getter( Settings.OPTION_smac2_count_tuner_time ) return self.__smac2_use_tunertime_in_cpu_time_budget @property def smac2_cli_cores(self: Settings) -> int: """Return the SMAC2 CLI cores.""" if self.__smac2_cli_cores is None: self.__smac2_cli_cores = self._abstract_getter( Settings.OPTION_smac2_cli_cores ) return self.__smac2_cli_cores @property def smac2_max_iterations(self: Settings) -> int: """Return the SMAC2 max iterations.""" if self.__smac2_max_iterations is None: self.__smac2_max_iterations = self._abstract_getter( Settings.OPTION_smac2_max_iterations ) return self.__smac2_max_iterations # SMAC3 attributes ### @property def smac3_number_of_trials(self: Settings) -> int: """Return the SMAC3 number of trials.""" if self.__smac3_number_of_trials is None: self.__smac3_number_of_trials = self._abstract_getter( Settings.OPTION_smac3_number_of_trials ) return self.__smac3_number_of_trials @property def smac3_facade(self: Settings) -> str: """Return the SMAC3 facade.""" if self.__smac3_facade is None: self.__smac3_facade = self._abstract_getter(Settings.OPTION_smac3_facade) return self.__smac3_facade @property def smac3_facade_max_ratio(self: Settings) -> float: """Return the SMAC3 facade max ratio.""" if self.__smac3_facade_max_ratio is None: self.__smac3_facade_max_ratio = self._abstract_getter( Settings.OPTION_smac3_facade_max_ratio ) return self.__smac3_facade_max_ratio @property def smac3_crash_cost(self: Settings) -> float: """Return the SMAC3 crash cost.""" if self.__smac3_crash_cost is None: self.__smac3_crash_cost = self._abstract_getter( Settings.OPTION_smac3_crash_cost ) return self.__smac3_crash_cost @property def smac3_termination_cost_threshold(self: Settings) -> float: """Return the SMAC3 termination cost threshold.""" if self.__smac3_termination_cost_threshold is None: self.__smac3_termination_cost_threshold = self._abstract_getter( Settings.OPTION_smac3_termination_cost_threshold ) return self.__smac3_termination_cost_threshold @property def smac3_wallclock_time_budget(self: Settings) -> int: """Return the SMAC3 walltime budget in seconds.""" if self.__smac3_wallclock_time_limit is None: self.__smac3_wallclock_time_limit = self._abstract_getter( Settings.OPTION_smac3_wallclock_time_budget ) return self.__smac3_wallclock_time_limit @property def smac3_cpu_time_budget(self: Settings) -> int: """Return the SMAC3 cputime budget in seconds.""" if self.__smac3_cputime_limit is None: self.__smac3_cputime_limit = self._abstract_getter( Settings.OPTION_smac3_cpu_time_budget ) return self.__smac3_cputime_limit @property def smac3_use_default_config(self: Settings) -> bool: """Return whether SMAC3 should use the default config.""" if self.__smac3_use_default_config is None: self.__smac3_use_default_config = self._abstract_getter( Settings.OPTION_smac3_use_default_config ) return self.__smac3_use_default_config @property def smac3_min_budget(self: Settings) -> int: """Return the SMAC3 min budget.""" if self.__smac3_min_budget is None: self.__smac3_min_budget = self._abstract_getter( Settings.OPTION_smac3_min_budget ) return self.__smac3_min_budget @property def smac3_max_budget(self: Settings) -> int: """Return the SMAC3 max budget.""" if self.__smac3_max_budget is None: self.__smac3_max_budget = self._abstract_getter( Settings.OPTION_smac3_max_budget ) return self.__smac3_max_budget # IRACE settings ### @property def irace_max_time(self: Settings) -> int: """Return the max time in seconds for IRACE.""" if self.__irace_max_time is None: self.__irace_max_time = self._abstract_getter(Settings.OPTION_irace_max_time) return self.__irace_max_time @property def irace_max_experiments(self: Settings) -> int: """Return the max experiments for IRACE.""" if self.__irace_max_experiments is None: self.__irace_max_experiments = self._abstract_getter( Settings.OPTION_irace_max_experiments ) return self.__irace_max_experiments @property def irace_first_test(self: Settings) -> int: """Return the first test for IRACE.""" if self.__irace_first_test is None: self.__irace_first_test = self._abstract_getter( Settings.OPTION_irace_first_test ) return self.__irace_first_test @property def irace_mu(self: Settings) -> int: """Return the mu for IRACE.""" if self.__irace_mu is None: self.__irace_mu = self._abstract_getter(Settings.OPTION_irace_mu) return self.__irace_mu @property def irace_max_iterations(self: Settings) -> int: """Return the max iterations for IRACE.""" if self.__irace_max_iterations is None: self.__irace_max_iterations = self._abstract_getter( Settings.OPTION_irace_max_iterations ) return self.__irace_max_iterations # ParamILS settings ### @property def paramils_cpu_time_budget(self: Settings) -> int: """Return the CPU time budget for ParamILS.""" if self.__paramils_cpu_time_budget is None: self.__paramils_cpu_time_budget = self._abstract_getter( Settings.OPTION_paramils_cpu_time_budget ) return self.__paramils_cpu_time_budget @property def paramils_min_runs(self: Settings) -> int: """Return the min runs for ParamILS.""" if self.__paramils_min_runs is None: self.__paramils_min_runs = self._abstract_getter( Settings.OPTION_paramils_min_runs ) return self.__paramils_min_runs @property def paramils_max_runs(self: Settings) -> int: """Return the max runs for ParamILS.""" if self.__paramils_max_runs is None: self.__paramils_max_runs = self._abstract_getter( Settings.OPTION_paramils_max_runs ) return self.__paramils_max_runs @property def paramils_random_restart(self: Settings) -> float: """Return the random restart for ParamILS.""" if self.__paramils_random_restart is None: self.__paramils_random_restart = self._abstract_getter( Settings.OPTION_paramils_random_restart ) return self.__paramils_random_restart @property def paramils_focused_approach(self: Settings) -> bool: """Return the focused approach for ParamILS.""" if self.__paramils_focused_approach is None: self.__paramils_focused_approach = self._abstract_getter( Settings.OPTION_paramils_focused ) return self.__paramils_focused_approach @property def paramils_use_cpu_time_in_tunertime(self: Settings) -> bool: """Return the use cpu time for ParamILS.""" if self.__paramils_use_cpu_time_in_tunertime is None: self.__paramils_use_cpu_time_in_tunertime = self._abstract_getter( Settings.OPTION_paramils_count_tuner_time ) return self.__paramils_use_cpu_time_in_tunertime @property def paramils_cli_cores(self: Settings) -> int: """The number of CPU cores to use for ParamILS.""" if self.__paramils_cli_cores is None: self.__paramils_cli_cores = self._abstract_getter( Settings.OPTION_paramils_cli_cores ) return self.__paramils_cli_cores @property def paramils_max_iterations(self: Settings) -> int: """Return the max iterations for ParamILS.""" if self.__paramils_max_iterations is None: self.__paramils_max_iterations = self._abstract_getter( Settings.OPTION_paramils_max_iterations ) return self.__paramils_max_iterations @property def paramils_number_initial_configurations(self: Settings) -> int: """Return the number of initial configurations for ParamILS.""" if self.__paramils_number_initial_configurations is None: self.__paramils_number_initial_configurations = self._abstract_getter( Settings.OPTION_paramils_number_initial_configurations ) return self.__paramils_number_initial_configurations # Parallel Portfolio settings ### @property def parallel_portfolio_check_interval(self: Settings) -> int: """Return the check interval for the parallel portfolio.""" if self.__parallel_portfolio_check_interval is None: self.__parallel_portfolio_check_interval = self._abstract_getter( Settings.OPTION_parallel_portfolio_check_interval ) return self.__parallel_portfolio_check_interval @property def parallel_portfolio_num_seeds_per_solver(self: Settings) -> int: """Return the number of seeds per solver for the parallel portfolio.""" if self.__parallel_portfolio_num_seeds_per_solver is None: self.__parallel_portfolio_num_seeds_per_solver = self._abstract_getter( Settings.OPTION_parallel_portfolio_number_of_seeds_per_solver ) return self.__parallel_portfolio_num_seeds_per_solver # Slurm settings ### @property def slurm_jobs_in_parallel(self: Settings) -> int: """Return the (maximum) number of jobs to run in parallel.""" if self.__slurm_jobs_in_parallel is None: self.__slurm_jobs_in_parallel = self._abstract_getter( Settings.OPTION_slurm_parallel_jobs ) return self.__slurm_jobs_in_parallel @property def slurm_job_prepend(self: Settings) -> str: """Return the slurm job prepend.""" if self.__slurm_job_prepend is None and self.__settings.has_option( Settings.OPTION_slurm_prepend_script.section, Settings.OPTION_slurm_prepend_script.name, ): value = self.__settings[Settings.OPTION_slurm_prepend_script.section][ Settings.OPTION_slurm_prepend_script.name ] try: path = Path(value) if path.is_file(): with path.open() as f: value = f.read() f.close() self.__slurm_job_prepend = str(value) except TypeError: pass return self.__slurm_job_prepend @property def sbatch_settings(self: Settings) -> list[str]: """Return the sbatch settings.""" sbatch_options = self.__settings[Settings.SECTION_slurm] # Return all non-predefined keys return [ f"--{key}={sbatch_options[key]}" for key in sbatch_options.keys() if key not in Settings.sections_options[Settings.SECTION_slurm] ] # General functionalities ###
[docs] def get_configurator_output_path(self: Settings, configurator: Configurator) -> Path: """Return the configurator output path.""" return self.DEFAULT_configuration_output / configurator.name
[docs] def get_configurator_settings( self: Settings, configurator_name: str ) -> dict[str, any]: """Return the settings of a specific configurator.""" configurator_settings = { "solver_calls": self.configurator_solver_call_budget, "solver_cutoff_time": self.solver_cutoff_time, "max_iterations": self.configurator_max_iterations, } # In the settings below, we default to the configurator general settings if no # specific configurator settings are given, by using the [None] or [Value] if configurator_name == cim.SMAC2.__name__: # Return all settings from the SMAC2 section configurator_settings.update( { "cpu_time": self.smac2_cpu_time_budget, "wallclock_time": self.smac2_wallclock_time_budget, "target_cutoff_length": self.smac2_target_cutoff_length, "use_cpu_time_in_tunertime": self.smac2_use_tunertime_in_cpu_time_budget, "cli_cores": self.smac2_cli_cores, "max_iterations": self.smac2_max_iterations or configurator_settings["max_iterations"], } ) elif configurator_name == cim.SMAC3.__name__: # Return all settings from the SMAC3 section del configurator_settings["max_iterations"] # SMAC3 does not have this? configurator_settings.update( { "smac_facade": self.smac3_facade, "max_ratio": self.smac3_facade_max_ratio, "crash_cost": self.smac3_crash_cost, "termination_cost_threshold": self.smac3_termination_cost_threshold, "walltime_limit": self.smac3_wallclock_time_budget, "cputime_limit": self.smac3_cpu_time_budget, "use_default_config": self.smac3_use_default_config, "min_budget": self.smac3_min_budget, "max_budget": self.smac3_max_budget, "solver_calls": self.smac3_number_of_trials or configurator_settings["solver_calls"], } ) # Do not pass None values to SMAC3, its Scenario resolves default settings configurator_settings = { key: value for key, value in configurator_settings.items() if value is not None } elif configurator_name == cim.IRACE.__name__: # Return all settings from the IRACE section configurator_settings.update( { "solver_calls": self.irace_max_experiments, "max_time": self.irace_max_time, "first_test": self.irace_first_test, "mu": self.irace_mu, "max_iterations": self.irace_max_iterations or configurator_settings["max_iterations"], } ) if ( configurator_settings["solver_calls"] == 0 and configurator_settings["max_time"] == 0 ): # Default to base configurator_settings["solver_calls"] = ( self.configurator_solver_call_budget ) elif configurator_name == cim.ParamILS.__name__: configurator_settings.update( { "tuner_timeout": self.paramils_cpu_time_budget, "min_runs": self.paramils_min_runs, "max_runs": self.paramils_max_runs, "focused_ils": self.paramils_focused_approach, "initial_configurations": self.paramils_number_initial_configurations, "random_restart": self.paramils_random_restart, "cli_cores": self.paramils_cli_cores, "use_cpu_time_in_tunertime": self.paramils_use_cpu_time_in_tunertime, "max_iterations": self.paramils_max_iterations or configurator_settings["max_iterations"], } ) return configurator_settings
[docs] @staticmethod def check_settings_changes( cur_settings: Settings, prev_settings: Settings, verbose: bool = True ) -> bool: """Check if there are changes between the previous and the current settings. Prints any section changes, printing None if no setting was found. Args: cur_settings: The current settings prev_settings: The previous settings verbose: Verbosity of the function Returns: True iff there are changes. """ cur_dict = cur_settings.__settings._sections prev_dict = prev_settings.__settings._sections cur_sections_set = set(cur_dict.keys()) prev_sections_set = set(prev_dict.keys()) sections_remained = cur_sections_set & prev_sections_set option_changed = False for section in sections_remained: printed_section = False names = set(cur_dict[section].keys()) | set(prev_dict[section].keys()) if ( section == "general" and "seed" in names ): # Do not report on the seed, is supposed to change names.remove("seed") for name in names: # if name is not present in one of the two dicts, get None as placeholder cur_val = cur_dict[section].get(name, None) prev_val = prev_dict[section].get(name, None) # If cur val is None, it is default if cur_val is not None and cur_val != prev_val: if not option_changed and verbose: # Print the initial print("[INFO] The following attributes/options have changed:") option_changed = True # do we have yet to print the section? if not printed_section and verbose: print(f" - In the section '{section}':") printed_section = True # print actual change if verbose: print(f" · '{name}' changed from '{prev_val}' to '{cur_val}'") return option_changed