Coverage for sparkle/CLI/generate_report.py: 89%
140 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-01-07 15:22 +0000
1#!/usr/bin/env python3
2"""Sparkle command to generate a report for an executed experiment."""
4import sys
5import argparse
6from pathlib import Path, PurePath
8from sparkle.CLI.help import global_variables as gv
9from sparkle.configurator.ablation import AblationScenario
10from sparkle.platform import generate_report_for_selection as sgfs
11from sparkle.platform import \
12 generate_report_for_configuration as sgrfch
13from sparkle.CLI.help import logging as sl
14from sparkle.platform.settings_objects import Settings, SettingState
15from sparkle.CLI.help import argparse_custom as ac
16from sparkle.CLI.help.reporting_scenario import Scenario
17from sparkle.platform import \
18 generate_report_for_parallel_portfolio as sgrfpph
19from sparkle.solver import Solver
20from sparkle.instance import Instance_Set
21from sparkle.structures import PerformanceDataFrame, FeatureDataFrame
22from sparkle.platform.output.configuration_output import ConfigurationOutput
23from sparkle.platform.output.selection_output import SelectionOutput
24from sparkle.platform.output.parallel_portfolio_output import ParallelPortfolioOutput
26from sparkle.CLI.initialise import check_for_initialise
27from sparkle.CLI.help.nicknames import resolve_object_name
30def parser_function() -> argparse.ArgumentParser:
31 """Define the command line arguments."""
32 parser = argparse.ArgumentParser(
33 description="Without any arguments a report for the most recent algorithm "
34 "selection or algorithm configuration procedure is generated.",
35 epilog="Note that if a test instance set is given, the training instance set "
36 "must also be given.")
37 # Configuration arguments
38 parser.add_argument(*ac.SolverReportArgument.names,
39 **ac.SolverReportArgument.kwargs)
40 parser.add_argument(*ac.InstanceSetTrainReportArgument.names,
41 **ac.InstanceSetTrainReportArgument.kwargs)
42 parser.add_argument(*ac.InstanceSetTestReportArgument.names,
43 **ac.InstanceSetTestReportArgument.kwargs)
44 parser.add_argument(*ac.NoAblationReportArgument.names,
45 **ac.NoAblationReportArgument.kwargs)
46 # Selection arguments
47 parser.add_argument(*ac.SelectionReportArgument.names,
48 **ac.SelectionReportArgument.kwargs)
49 parser.add_argument(*ac.TestCaseDirectoryArgument.names,
50 **ac.TestCaseDirectoryArgument.kwargs)
51 # Common arguments
52 parser.add_argument(*ac.ObjectivesArgument.names,
53 **ac.ObjectivesArgument.kwargs)
54 parser.add_argument(*ac.SettingsFileArgument.names,
55 **ac.SettingsFileArgument.kwargs)
56 parser.add_argument(*ac.GenerateJSONArgument.names,
57 **ac.GenerateJSONArgument.kwargs)
58 return parser
61def main(argv: list[str]) -> None:
62 """Generate a report for an executed experiment."""
63 # Log command call
64 sl.log_command(sys.argv)
65 check_for_initialise()
67 # Define command line arguments
68 parser = parser_function()
70 # Process command line arguments
71 args = parser.parse_args(argv)
73 # Do first, so other command line options can override settings from the file
74 if ac.set_by_user(args, "settings_file"):
75 gv.settings().read_settings_ini(
76 args.settings_file, SettingState.CMD_LINE
77 )
78 if args.objectives is not None:
79 gv.settings().set_general_sparkle_objectives(
80 args.objectives, SettingState.CMD_LINE)
81 selection = args.selection
82 test_case_dir = args.test_case_directory
83 only_json = args.only_json
85 solver = resolve_object_name(
86 args.solver,
87 gv.file_storage_data_mapping[gv.solver_nickname_list_path],
88 gv.settings().DEFAULT_solver_dir, Solver)
89 instance_set_train = resolve_object_name(
90 args.instance_set_train,
91 gv.file_storage_data_mapping[gv.instances_nickname_path],
92 gv.settings().DEFAULT_instance_dir, Instance_Set)
93 instance_set_test = resolve_object_name(
94 args.instance_set_train,
95 gv.file_storage_data_mapping[gv.instances_nickname_path],
96 gv.settings().DEFAULT_instance_dir, Instance_Set)
98 Settings.check_settings_changes(gv.settings(),
99 Settings(PurePath("Settings/latest.ini")))
100 # If no arguments are set get the latest scenario
101 if not selection and test_case_dir is None and solver is None:
102 scenario = gv.latest_scenario().get_latest_scenario()
103 if scenario == Scenario.SELECTION:
104 selection = True
105 test_case_dir = gv.latest_scenario().get_selection_test_case_directory()
106 elif scenario == Scenario.CONFIGURATION:
107 solver = gv.latest_scenario().get_config_solver()
108 instance_set_train = gv.latest_scenario().get_config_instance_set_train()
109 instance_set_test = gv.latest_scenario().get_config_instance_set_test()
110 elif scenario == Scenario.PARALLEL_PORTFOLIO:
111 parallel_portfolio_path = gv.latest_scenario().get_parallel_portfolio_path()
112 pap_instance_set =\
113 gv.latest_scenario().get_parallel_portfolio_instance_set()
115 flag_instance_set_train = instance_set_train is not None
116 flag_instance_set_test = instance_set_test is not None
118 # Reporting for algorithm selection
119 if selection or test_case_dir is not None:
120 objective = gv.settings().get_general_sparkle_objectives()[0]
121 if not objective.time:
122 print("ERROR: The selection report is not implemented for "
123 " non-runtime objectives!")
124 sys.exit(-1)
125 selection_scenario = gv.latest_scenario().get_selection_scenario_path()
126 actual_portfolio_selector_path = selection_scenario / "portfolio_selector"
127 if not actual_portfolio_selector_path.is_file():
128 print("Before generating a Sparkle report, please first construct the "
129 "Sparkle portfolio selector. Not generating a Sparkle report, stopping"
130 " execution!")
131 sys.exit(-1)
133 print("Generating report for selection...")
134 train_data = PerformanceDataFrame(gv.settings().DEFAULT_performance_data_path)
135 feature_data = FeatureDataFrame(gv.settings().DEFAULT_feature_data_path)
136 test_data = None
137 test_case_path = Path(test_case_dir) if test_case_dir is not None else None
138 if test_case_dir is not None and (test_case_path
139 / "performance_data.csv").exists():
140 test_data = PerformanceDataFrame(test_case_path / "performance_data.csv")
141 # Create machine readable selection output
142 instance_dirs = set(Path(instance).parent for instance in train_data.instances)
143 instance_sets = []
144 for dir in instance_dirs:
145 instance_sets.append(Instance_Set(dir))
146 test_set = None if test_case_dir is None else Instance_Set(Path(test_case_dir))
147 cutoff_time = gv.settings().get_general_target_cutoff_time()
148 output = gv.settings().DEFAULT_selection_output_analysis
149 selection_output = SelectionOutput(
150 selection_scenario, train_data, feature_data,
151 instance_sets, test_set, objective, cutoff_time,
152 output)
153 selection_output.write_output()
154 print("Machine readable output is placed at:", selection_output.output)
156 if not only_json:
157 sgfs.generate_report_selection(
158 gv.settings().DEFAULT_selection_output_analysis,
159 gv.settings().DEFAULT_latex_source,
160 "template-Sparkle-for-selection.tex",
161 gv.settings().DEFAULT_latex_bib,
162 gv.settings().DEFAULT_extractor_dir,
163 selection_scenario,
164 feature_data,
165 train_data,
166 objective,
167 gv.settings().get_general_extractor_cutoff_time(),
168 gv.settings().get_general_target_cutoff_time(),
169 test_data
170 )
171 if test_case_dir is None:
172 print("Report generated ...")
173 else:
174 print("Report for test generated ...")
176 elif gv.latest_scenario().get_latest_scenario() == Scenario.PARALLEL_PORTFOLIO:
177 # Reporting for parallel portfolio
178 # Machine readable Output
179 cutoff_time = gv.settings().get_general_target_cutoff_time()
180 objective = gv.settings().get_general_sparkle_objectives()[0]
181 output = gv.settings().DEFAULT_parallel_portfolio_output_analysis
182 parallel_portfolio_output = ParallelPortfolioOutput(parallel_portfolio_path,
183 pap_instance_set,
184 objective,
185 output)
186 parallel_portfolio_output.write_output()
187 print("Machine readable output is placed at:", parallel_portfolio_output.output)
189 if not only_json:
190 sgrfpph.generate_report_parallel_portfolio(
191 parallel_portfolio_path,
192 gv.settings().DEFAULT_parallel_portfolio_output_analysis,
193 gv.settings().DEFAULT_latex_source,
194 gv.settings().DEFAULT_latex_bib,
195 gv.settings().get_general_sparkle_objectives()[0],
196 gv.settings().get_general_target_cutoff_time(),
197 pap_instance_set)
198 print("Parallel portfolio report generated ...")
199 else:
200 # Reporting for algorithm configuration
201 if solver is None:
202 print("Error! No Solver found for configuration report generation.")
203 sys.exit(-1)
205 # If only the testing set is given return an error
206 if not flag_instance_set_train and flag_instance_set_test:
207 print("Argument Error! Only a testing set was provided, please also "
208 "provide a training set")
209 print(f"Usage: {sys.argv[0]} --solver <solver> [--instance-set-train "
210 "<instance-set-train>] [--instance-set-test <instance-set-test>]")
211 sys.exit(-1)
212 # Extract config scenario data for report, but this should be read from the
213 # scenario file instead as we can't know wether features were used or not now
214 configurator = gv.settings().get_general_sparkle_configurator()
215 config_scenario = gv.latest_scenario().get_configuration_scenario(
216 configurator.scenario_class())
217 ablation_scenario = None
218 if args.flag_ablation:
219 ablation_scenario = AblationScenario(
220 config_scenario, instance_set_test,
221 gv.settings().DEFAULT_ablation_output)
222 performance_data = PerformanceDataFrame(
223 gv.settings().DEFAULT_performance_data_path)
224 # Filter the performance data to solver/instance sets
225 performance_data = performance_data
226 if performance_data.num_solvers > 1:
227 performance_data.remove_solver(
228 [s for s in performance_data.solvers
229 if s != solver.directory])
231 used_instances = [str(i) for i in instance_set_train.instance_paths]
232 if instance_set_test:
233 used_instances += [str(i) for i in instance_set_test.instance_paths]
234 for i in performance_data.instances:
235 if i not in used_instances:
236 performance_data.remove_instance(i)
237 configuration_jobs = performance_data.get_job_list()
238 if len(configuration_jobs) > 0:
239 print(f"ERROR: {(len(configuration_jobs))} jobs for the configuration were "
240 "not executed! Please run 'sparkle run solvers --performance-data' "
241 "before continuing.")
242 sys.exit(-1)
243 # Create machine readable output
244 output = gv.settings().DEFAULT_configuration_output_analysis
245 config_output = ConfigurationOutput(config_scenario.directory,
246 configurator,
247 config_scenario,
248 performance_data,
249 instance_set_test,
250 output)
251 config_output.write_output()
252 print("Machine readable output is placed at:", config_output.output)
254 if not only_json:
255 sgrfch.generate_report_for_configuration(
256 config_scenario,
257 config_output,
258 gv.settings().DEFAULT_extractor_dir,
259 gv.settings().DEFAULT_configuration_output_analysis,
260 gv.settings().DEFAULT_latex_source,
261 gv.settings().DEFAULT_latex_bib,
262 gv.settings().get_general_extractor_cutoff_time(),
263 ablation=ablation_scenario,
264 )
266 # Write used settings to file
267 gv.settings().write_used_settings()
268 sys.exit(0)
271if __name__ == "__main__":
272 main(sys.argv[1:])