Coverage for src/sparkle/platform/output/configuration_output.py: 93%
55 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
5from sparkle.structures import PerformanceDataFrame
6from sparkle.instance import InstanceSet
7from sparkle.configurator.configurator import ConfigurationScenario
8from sparkle.types import SparkleObjective, SolverStatus
11class ConfigurationResult:
12 """Class that represents result of configuration on an instance set."""
14 def __init__(
15 self: ConfigurationResult,
16 instance_set: str,
17 default_instance_performance: list[float],
18 best_instance_performance: list[float],
19 instance_status_default: dict[str, SolverStatus],
20 instance_status_best: dict[str, SolverStatus],
21 objective: SparkleObjective,
22 ) -> None:
23 """Initialize a configuration result.
25 All input sequences are the results per instance.
27 Args:
28 instance_set: The name of the instance set
29 default_instance_performance: The default instance performance
30 best_instance_performance: The best instance performance
31 performance: The performance of the configuration
32 instance_status_default: The status of the default configuration
33 instance_status_best: The status of the best configuration
34 objective: The objective
35 """
36 self.default_instance_performance = default_instance_performance
37 self.default_performance: float = objective.instance_aggregator(
38 default_instance_performance
39 )
40 self.best_instance_performance = best_instance_performance
41 self.best_performance: float = objective.instance_aggregator(
42 best_instance_performance
43 )
44 self.instance_status_default = instance_status_default
45 self.instance_status_best = instance_status_best
46 self.instance_set_name = instance_set
48 def serialise(self: ConfigurationResult) -> dict[str, float | list[float]]:
49 """Serialise the data."""
50 return {
51 "instance_set": self.instance_set_name,
52 "default_performance": self.default_performance,
53 "best_performance": self.best_performance,
54 "default_instance_performance": self.default_instance_performance,
55 "best_instance_performance": self.best_instance_performance,
56 "instance_status_default": self.instance_status_default,
57 "instance_status_best": self.instance_status_best,
58 }
61class ConfigurationOutput:
62 """Class that collects configuration data and outputs it a JSON format."""
64 def __init__(
65 self: ConfigurationOutput,
66 config_scenario: ConfigurationScenario,
67 performance_data: PerformanceDataFrame,
68 possible_test_sets: list[InstanceSet] = None,
69 ) -> None:
70 """Initialize Configurator Output class.
72 Args:
73 config_scenario: The scenario to output
74 performance_data: Performance data
75 possible_test_sets: Instance Sets possibly used for testing
76 """
77 self.solver = config_scenario.solver
78 self.configurator = config_scenario.configurator
79 self.instance_set_train = config_scenario.instance_set
81 # Filter data on this scenario
82 performance_data_config = performance_data.clone()
83 performance_data_config.remove_solver(
84 [
85 solver
86 for solver in performance_data_config.solvers
87 if solver != str(self.solver.directory)
88 ]
89 )
90 used_configs = config_scenario.configuration_ids + [
91 PerformanceDataFrame.default_configuration
92 ]
93 removable = [
94 configuration
95 for configuration in performance_data_config.configuration_ids
96 if configuration not in used_configs
97 ]
98 performance_data_config.remove_configuration(
99 str(self.solver.directory), removable
100 )
101 self.test_instance_sets = []
102 for test_set in possible_test_sets:
103 if test_set.name == self.instance_set_train.name:
104 continue
105 for instance_set, instance_name in test_set.instance_pairs:
106 if (
107 instance_set,
108 instance_name,
109 ) not in performance_data_config.instance_pairs or (
110 performance_data_config.is_missing(
111 str(self.solver.directory), instance_set, instance_name
112 )
113 ):
114 break
115 else:
116 # Only keep test sets whose instances all have performance data
117 self.test_instance_sets.append(test_set)
118 self.directory = config_scenario.directory
119 self.config_scenario = config_scenario
121 # Retrieve all configurations
122 solver_key = str(self.solver.directory)
123 config_keys = performance_data_config.get_configurations(solver_key)
124 self.all_configurations = performance_data_config.get_full_configuration(
125 solver_key, config_keys
126 )
128 # Retrieve configuration performances
129 train_instance_pairs = self.instance_set_train.instance_pairs
130 # Retrieve Default (No configuration) performance
131 _, self.default_performance_train = (
132 performance_data_config.configuration_performance(
133 solver_key,
134 PerformanceDataFrame.default_configuration,
135 objective=self.config_scenario.sparkle_objectives[0],
136 instance_pairs=train_instance_pairs,
137 )
138 )
140 _, self.default_performance_per_instance_train = (
141 performance_data_config.configuration_performance(
142 solver_key,
143 PerformanceDataFrame.default_configuration,
144 objective=self.config_scenario.sparkle_objectives[0],
145 instance_pairs=train_instance_pairs,
146 per_instance=True,
147 )
148 )
150 # Retrieve best found configuration
151 self.best_configuration_key, self.best_performance_train = (
152 performance_data_config.best_configuration(
153 solver_key,
154 objective=self.config_scenario.sparkle_objective,
155 instance_pairs=train_instance_pairs,
156 )
157 )
158 self.best_configuration = self.all_configurations[
159 config_keys.index(self.best_configuration_key)
160 ]
162 # TODO keep all instance set performance data together in a dictionary instead
163 # of variables for train and test
164 # Shitty hack to get status objective
165 status_objective = [
166 objective
167 for objective in performance_data_config.objective_names
168 if objective.lower().startswith("status")
169 ][0]
170 self.instance_set_results: dict[str, ConfigurationResult] = {}
171 for instance_set in self.test_instance_sets + [self.instance_set_train]:
172 instance_pairs = instance_set.instance_pairs
173 _, default_performance_per_instance = (
174 performance_data_config.configuration_performance(
175 solver_key,
176 PerformanceDataFrame.default_configuration,
177 objective=self.config_scenario.sparkle_objective,
178 instance_pairs=instance_pairs,
179 per_instance=True,
180 )
181 )
182 _, best_conf_performance_per_instance = (
183 performance_data_config.configuration_performance(
184 solver_key,
185 self.best_configuration_key,
186 objective=self.config_scenario.sparkle_objective,
187 instance_pairs=instance_pairs,
188 per_instance=True,
189 )
190 )
191 instance_status_default = {
192 instance_name: performance_data_config.get_value(
193 solver_key,
194 configuration=PerformanceDataFrame.default_configuration,
195 objective=status_objective,
196 instance_pair=(instance_set_name, instance_name),
197 )
198 for instance_set_name, instance_name in instance_pairs
199 }
200 instance_status_best_conf = {
201 instance_name: performance_data_config.get_value(
202 solver_key,
203 configuration=self.best_configuration_key,
204 objective=status_objective,
205 instance_pair=(instance_set_name, instance_name),
206 )
207 for instance_set_name, instance_name in instance_pairs
208 }
209 self.instance_set_results[instance_set.name] = ConfigurationResult(
210 instance_set.name,
211 default_performance_per_instance,
212 best_conf_performance_per_instance,
213 instance_status_default,
214 instance_status_best_conf,
215 self.config_scenario.sparkle_objectives[0],
216 )
218 def serialise(self: ConfigurationOutput) -> dict:
219 """Serialise the configuration output."""
220 return {
221 "solver": self.solver.name,
222 "configurator": self.configurator.__name__,
223 "best_configuration": self.best_configuration,
224 "best_performance_train": self.best_performance_train,
225 "scenario": {
226 str(key): str(value)
227 for key, value in self.config_scenario.serialise().items()
228 },
229 }