Coverage for src/sparkle/platform/output/selection_output.py: 73%
75 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-08 12:00 +0000
1"""Sparkle class to organise configuration output."""
3from __future__ import annotations
4import operator
5import json
6from pathlib import Path
8from sparkle.selector import SelectionScenario
9from sparkle.structures import PerformanceDataFrame
10from sparkle.platform.output.structures import (
11 SelectionPerformance,
12 SelectionSolverData,
13)
16def compute_selector_marginal_contribution(
17 selection_scenario: SelectionScenario,
18) -> list[tuple[str, float]]:
19 """Compute the marginal contributions of solvers in the selector.
21 Args:
22 performance_data: Performance data object
23 feature_data_csv_path: Path to the CSV file with the feature data.
24 selection_scenario: The selector scenario for which to compute
25 marginal contribution.
26 objective: Objective to compute the marginal contribution for.
28 Returns:
29 A list of 4-tuples where every 4-tuple is of the form
30 (solver_name, config_id, marginal contribution, best_performance).
31 """
32 selector_performance = selection_scenario.objective.instance_aggregator(
33 selection_scenario.selector_performance_data.get_value(
34 SelectionScenario.__selector_solver_name__,
35 instance_pair=selection_scenario.training_instances,
36 objective=selection_scenario.objective.name,
37 )
38 )
39 rank_list = []
40 compare = operator.lt if selection_scenario.objective.minimise else operator.gt
41 # Compute contribution per solver
42 for ablation_scenario in selection_scenario.ablation_scenarios:
43 # Hacky way of getting the needed data on the ablation
44 _, solver_name, config = ablation_scenario.directory.name.split("_", maxsplit=2)
45 # Hacky way of reconstructing the solver id in the PDF
46 solver = f"Solvers/{solver_name}"
47 ablated_selector_performance = ablation_scenario.objective.instance_aggregator(
48 ablation_scenario.selector_performance_data.get_value(
49 SelectionScenario.__selector_solver_name__,
50 instance_pair=ablation_scenario.training_instances,
51 objective=ablation_scenario.objective.name,
52 )
53 )
55 # 1. If the performance remains equal, this solver did not contribute
56 # 2. If there is a performance decay without this solver, it does contribute
57 # 3. If there is a performance improvement, we have a bad portfolio selector
58 if ablated_selector_performance == selector_performance:
59 marginal_contribution = 0.0
60 elif not compare(ablated_selector_performance, selector_performance):
61 # The performance decreases, we have a contributing solver
62 marginal_contribution = ablated_selector_performance / selector_performance
63 else:
64 print(
65 "****** WARNING DUBIOUS SELECTOR/SOLVER: "
66 f"The omission of solver {solver_name} ({config}) yields an "
67 "improvement. The selector improves better without this solver. "
68 "It may be usefull to construct a portfolio without this solver."
69 )
70 marginal_contribution = 0.0
72 rank_list.append(
73 (solver, config, marginal_contribution, ablated_selector_performance)
74 )
76 rank_list.sort(key=lambda contribution: contribution[2], reverse=True)
77 return rank_list
80class SelectionOutput:
81 """Class that collects selection data and outputs it a JSON format."""
83 def __init__(
84 self: SelectionOutput,
85 selection_scenario: SelectionScenario,
86 ) -> None:
87 """Initialize SelectionOutput class.
89 Args:
90 selection_scenario: Path to selection output directory
91 performance_data: The performance data used for the selector
92 """
93 self.training_instances = selection_scenario.training_instances
94 training_instance_sets = selection_scenario.training_instance_sets
95 self.training_instance_sets = [
96 (
97 instance_set,
98 sum(inst_set == instance_set for inst_set, _ in self.training_instances),
99 )
100 for instance_set in training_instance_sets
101 ]
102 self.test_instances = selection_scenario.test_instances
103 test_sets = selection_scenario.test_instance_sets
104 self.test_sets = [
105 (
106 instance_set,
107 sum(inst_set == instance_set for inst_set, _ in self.test_instances),
108 )
109 for instance_set in test_sets
110 ]
111 self.cutoff_time = selection_scenario.solver_cutoff
112 self.objective = selection_scenario.objective
114 solver_performance_data = selection_scenario.selector_performance_data.clone()
115 solver_performance_data.remove_solver(SelectionScenario.__selector_solver_name__)
117 self.solver_performance_ranking = solver_performance_data.get_solver_ranking(
118 instance_pairs=self.training_instances, objective=self.objective
119 )
121 self.solver_data = self.get_solver_data(solver_performance_data)
122 self.solvers = {}
123 for solver_conf in selection_scenario.performance_data.columns:
124 solver, conf = solver_conf.split("_", maxsplit=1)
125 if solver not in self.solvers:
126 self.solvers[solver] = []
127 self.solvers[solver].append(conf)
129 self.sbs_performance = solver_performance_data.get_value(
130 solver=self.solver_performance_ranking[0][0],
131 configuration=self.solver_performance_ranking[0][1],
132 instance_pair=self.training_instances,
133 objective=self.objective.name,
134 )
136 # Collect marginal contribution data
137 self.marginal_contribution_perfect = (
138 solver_performance_data.marginal_contribution(
139 selection_scenario.objective,
140 instance_pairs=self.training_instances,
141 sort=True,
142 )
143 )
145 self.marginal_contribution_actual = compute_selector_marginal_contribution(
146 selection_scenario
147 )
148 # Collect performance data
149 self.vbs_performance_data = solver_performance_data.best_instance_performance(
150 instance_pairs=self.training_instances,
151 objective=selection_scenario.objective,
152 )
153 self.vbs_performance = selection_scenario.objective.instance_aggregator(
154 self.vbs_performance_data
155 )
157 self.test_set_performance = {} if self.test_sets else None
158 for test_set, _ in self.test_sets:
159 test_set_instances = [
160 instance for instance in self.test_instances if instance[0] == test_set
161 ]
162 test_perf = selection_scenario.selector_performance_data.best_performance(
163 exclude_solvers=[
164 solver
165 for solver in selection_scenario.selector_performance_data.solvers
166 if solver != SelectionScenario.__selector_solver_name__
167 ],
168 instance_pairs=test_set_instances,
169 objective=selection_scenario.objective,
170 )
171 self.test_set_performance[test_set] = test_perf
172 self.actual_performance_data = (
173 selection_scenario.selector_performance_data.get_value(
174 solver=SelectionScenario.__selector_solver_name__,
175 instance_pair=self.training_instances,
176 objective=self.objective.name,
177 )
178 )
179 self.actual_performance = self.objective.instance_aggregator(
180 self.actual_performance_data
181 )
183 def get_solver_data(
184 self: SelectionOutput, train_data: PerformanceDataFrame
185 ) -> SelectionSolverData:
186 """Initalise SelectionSolverData object."""
187 num_solvers = train_data.num_solvers
188 return SelectionSolverData(self.solver_performance_ranking, num_solvers)
190 def serialise_solvers(self: SelectionOutput, sd: SelectionSolverData) -> dict:
191 """Transform SelectionSolverData to dictionary format."""
192 return {
193 "number_of_solvers": sd.num_solvers,
194 "single_best_solver": sd.single_best_solver,
195 "solver_ranking": [
196 {"solver_name": solver[0], "performance": solver[1]}
197 for solver in sd.solver_performance_ranking
198 ],
199 }
201 def serialise_performance(self: SelectionOutput, sp: SelectionPerformance) -> dict:
202 """Transform SelectionPerformance to dictionary format."""
203 return {
204 "vbs_performance": sp.vbs_performance,
205 "actual_performance": sp.actual_performance,
206 "objective": self.objective.name,
207 "metric": sp.metric,
208 }
210 def serialise_instances(
211 self: SelectionOutput, instance_pairs: list[tuple[str, str]]
212 ) -> dict:
213 """Transform Instances to dictionary format."""
214 instance_sets = dict.fromkeys(instance_set for instance_set, _ in instance_pairs)
215 return {
216 "number_of_instance_sets": len(instance_sets),
217 "instance_sets": [
218 {
219 "name": instance_set,
220 "number_of_instances": len(
221 [
222 inst_set
223 for inst_set, _ in instance_pairs
224 if inst_set == instance_set
225 ]
226 ),
227 }
228 for instance_set in instance_sets
229 ],
230 }
232 def serialise_marginal_contribution(self: SelectionOutput) -> dict:
233 """Transform performance ranking to dictionary format."""
234 return {
235 "marginal_contribution_actual": [
236 {
237 "solver_name": ranking[0],
238 "marginal_contribution": ranking[1],
239 "best_performance": ranking[2],
240 }
241 for ranking in self.marginal_contribution_actual
242 ],
243 "marginal_contribution_perfect": [
244 {
245 "solver_name": ranking[0],
246 "marginal_contribution": ranking[1],
247 "best_performance": ranking[2],
248 }
249 for ranking in self.marginal_contribution_perfect
250 ],
251 }
253 def serialise(self: SelectionOutput) -> dict:
254 """Serialise the selection output."""
255 test_data = (
256 self.serialise_instances(self.test_instances)
257 if self.test_instances
258 else None
259 )
260 return {
261 "solvers": self.serialise_solvers(self.solver_data),
262 "training_instances": self.serialise_instances(self.training_instances),
263 "test_instances": test_data,
264 "settings": {"cutoff_time": self.cutoff_time},
265 "marginal_contribution": self.serialise_marginal_contribution(),
266 }
268 def write_output(self: SelectionOutput, output: Path) -> None:
269 """Write data into a JSON file."""
270 output = output / "configuration.json" if output.is_dir() else output
271 with output.open("w") as f:
272 json.dump(self.serialise(), f, indent=4)