Coverage for sparkle/CLI/initialise.py: 76%
108 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-09-29 10:17 +0000
1#!/usr/bin/env python3
2"""Command to initialise a Sparkle platform."""
4import subprocess
5import argparse
6import shutil
7import os
8import sys
9import warnings
10from pathlib import Path
12from sparkle.CLI.help import argparse_custom as ac
13from sparkle.CLI.help import snapshot_help as snh
14from sparkle.CLI.help import global_variables as gv
15from sparkle.configurator.implementations import IRACE, SMAC2, ParamILS
16from sparkle.platform import Settings
17from sparkle.structures import PerformanceDataFrame, FeatureDataFrame
20def parser_function() -> argparse.ArgumentParser:
21 """Parse CLI arguments for the initialise command."""
22 parser = argparse.ArgumentParser(
23 description="Initialise the Sparkle platform in the current directory."
24 )
25 parser.add_argument(
26 *ac.DownloadExamplesArgument.names, **ac.DownloadExamplesArgument.kwargs
27 )
28 parser.add_argument(
29 *ac.NoSavePlatformArgument.names, **ac.NoSavePlatformArgument.kwargs
30 )
31 parser.add_argument(
32 *ac.RebuildRunsolverArgument.names, **ac.RebuildRunsolverArgument.kwargs
33 )
34 return parser
37def detect_sparkle_platform_exists(check: callable = all) -> Path:
38 """Return whether a Sparkle platform is currently active.
40 The default working directories are checked for existence, for each directory in the
41 CWD. If any of the parents of the CWD yield true, this path is returned
43 Args:
44 check: Method to check if the working directory exists. Defaults to all.
46 Returns:
47 Path to the Sparkle platform if it exists, None otherwise.
48 """
49 cwd = Path.cwd()
50 while str(cwd) != cwd.root:
51 if check([(cwd / wd).exists() for wd in Settings.DEFAULT_working_dirs]):
52 return cwd
53 cwd = cwd.parent
54 return None
57def check_for_initialise() -> None:
58 """Function to check if initialize command was executed and execute it otherwise.
60 Args:
61 argv: List of the arguments from the caller.
62 requirements: The requirements that have to be executed before the calling
63 function.
64 """
65 platform_path = detect_sparkle_platform_exists()
66 if platform_path is None:
67 print("-----------------------------------------------")
68 print(
69 "No Sparkle platform found; "
70 "The platform will now be initialized automatically."
71 )
72 print("-----------------------------------------------")
73 initialise_sparkle()
74 elif platform_path != Path.cwd():
75 print(
76 f"[WARNING] Sparkle platform found in {platform_path} instead of "
77 f"{Path.cwd()}. Switching to CWD to {platform_path}"
78 )
79 os.chdir(platform_path)
82def initialise_sparkle(
83 save_existing_platform: bool = True,
84 interactive: bool = False,
85 download_examples: bool = False,
86 rebuild_runsolver: bool = False,
87) -> None:
88 """Initialize a new Sparkle platform.
90 Args:
91 save_existing_platform: If present, save the current platform as a snapshot.
92 interactive: Ask for user input or not.
93 download_examples: Downloads examples from the Sparkle Github.
94 WARNING: May take a some time to complete due to the large amount of data.
95 rebuild_runsolver: Will clean the RunSolver executable and rebuild it.
96 """
97 print("Start initialising Sparkle platform ...")
98 if detect_sparkle_platform_exists(check=all):
99 print("Current Sparkle platform found!")
100 if save_existing_platform:
101 print("Saving as snapshot...")
102 snh.save_current_platform()
103 snh.remove_current_platform(filter=[Settings.DEFAULT_settings_dir])
104 print("Your settings directory was not removed.")
106 for working_dir in Settings.DEFAULT_working_dirs:
107 working_dir.mkdir(exist_ok=True)
109 # Check if Settings file exists, otherwise initialise a default one
110 if not Path(Settings.DEFAULT_settings_path).exists():
111 print("Settings file does not exist, initializing default settings ...")
112 gv.__settings = Settings(Settings.DEFAULT_example_settings_path)
113 gv.settings().write_settings_ini(Path(Settings.DEFAULT_settings_path))
115 # Initialise the FeatureDataFrame
116 FeatureDataFrame(Settings.DEFAULT_feature_data_path)
118 # Initialise the Performance DF with the static dimensions
119 # TODO: We have many sparkle settings values regarding ``number of runs''
120 # E.g. configurator, parallel portfolio, and here too. Should we unify this more, or
121 # just make another setting that does this specifically for performance data?
122 PerformanceDataFrame(
123 Settings.DEFAULT_performance_data_path,
124 objectives=gv.settings().objectives,
125 n_runs=1,
126 )
128 if rebuild_runsolver:
129 print("Cleaning Runsolver ...")
130 runsolver_clean = subprocess.run(
131 ["make", "clean"], cwd=Settings.DEFAULT_runsolver_dir, capture_output=True
132 )
133 if runsolver_clean.returncode != 0:
134 warnings.warn(
135 f"[{runsolver_clean.returncode}] Cleaning of Runsolver failed "
136 f"with the following msg: {runsolver_clean.stdout.decode()}"
137 )
139 # Check that Runsolver is compiled, otherwise, compile
140 if not Settings.DEFAULT_runsolver_exec.exists():
141 print("Runsolver does not exist, trying to compile...")
142 if not (Settings.DEFAULT_runsolver_dir / "Makefile").exists():
143 warnings.warn(
144 "Runsolver executable doesn't exist and cannot find makefile."
145 " Please verify the contents of the directory: "
146 f"{Settings.DEFAULT_runsolver_dir}"
147 )
148 else:
149 compile_runsolver = subprocess.run(
150 ["make"], cwd=Settings.DEFAULT_runsolver_dir, capture_output=True
151 )
152 if compile_runsolver.returncode != 0:
153 warnings.warn(
154 "Compilation of Runsolver failed with the following msg:"
155 f"[{compile_runsolver.returncode}] "
156 f"{compile_runsolver.stderr.decode()}"
157 )
158 else:
159 print("Runsolver compiled successfully!")
161 # If Runsolver is compiled, check that it can be executed
162 if Settings.DEFAULT_runsolver_exec.exists():
163 runsolver_check = subprocess.run(
164 [Settings.DEFAULT_runsolver_exec, "--version"], capture_output=True
165 )
166 if runsolver_check.returncode != 0:
167 print(
168 "WARNING: Runsolver executable cannot be run successfully. "
169 "Please verify the following error messages:\n"
170 f"{runsolver_check.stderr.decode()}"
171 )
173 # Check that java is available for SMAC2
174 if shutil.which("java") is None:
175 # NOTE: An automatic resolution of Java at this point would be good
176 # However, loading modules from Python has thusfar not been successfull.
177 print(
178 "WARNING: Could not find Java as an executable! Java 1.8.0_402 is required"
179 " to use SMAC2 or ParamILS as a configurator. Consider installing Java."
180 )
182 # Check for each configurator that it is available
183 if interactive and not SMAC2.configurator_executable.exists():
184 print("SMAC2 is not installed, would you like to install? (Y/n) ...")
185 if input().lower() == "y":
186 print("Installing SMAC2 ...")
187 SMAC2.download_requirements()
188 if interactive and not ParamILS.configurator_executable.exists():
189 print("ParamILS is not installed, would you like to install? (Y/n) ...")
190 if input().lower() == "y":
191 print("Installing ParamILS ...")
192 ParamILS.download_requirements()
193 if interactive and not IRACE.check_requirements():
194 if shutil.which("R") is None:
195 print(
196 "R is not installed, which is required for the IRACE "
197 "configurator (installation). Consider installing R."
198 )
199 else:
200 print("IRACE is not installed, would you like to install? (Y/n) ...")
201 if input().lower() == "y":
202 print("Installing IRACE ...")
203 IRACE.download_requirements()
204 irace = IRACE()
205 print(f"Installed IRACE version {irace.version}")
207 if download_examples:
208 # Download Sparkle examples from Github
209 print("Downloading examples ...")
210 curl = subprocess.Popen(
211 ["curl", "https://codeload.github.com/ADA-research/Sparkle/tar.gz/main"],
212 stdout=subprocess.PIPE,
213 )
214 outpath = Path("outfile.tar.gz")
215 with curl.stdout, outpath.open("wb") as outfile:
216 tar = subprocess.Popen(
217 ["tar", "-xz", "--strip=1", "Sparkle-main/Examples"],
218 stdin=curl.stdout,
219 stdout=outfile,
220 )
221 curl.wait() # Wait for the download to complete
222 tar.wait() # Wait for the extraction to complete
223 outpath.unlink(missing_ok=True)
225 print("New Sparkle platform initialised!")
228def main(argv: list[str]) -> None:
229 """Main function of the command."""
230 # Define command line arguments
231 parser = parser_function()
232 # Process command line arguments
233 args = parser.parse_args(argv)
234 initialise_sparkle(
235 save_existing_platform=args.no_save,
236 interactive=True,
237 download_examples=args.download_examples,
238 rebuild_runsolver=args.rebuild_runsolver,
239 )
240 sys.exit(0)
243if __name__ == "__main__":
244 main(sys.argv[1:])