Coverage for src/sparkle/CLI/cleanup.py: 27%
137 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-08 12:00 +0000
1#!/usr/bin/env python3
2"""Command to remove temporary files not affecting the platform state."""
4import re
5import math
6import sys
7import argparse
8import shutil
10from sparkle.structures import PerformanceDataFrame, FeatureDataFrame
11from sparkle.types import DataFileLock
13from sparkle.CLI.help import logging as sl
14from sparkle.CLI.help import global_variables as gv
15from sparkle.CLI.help import argparse_custom as ac
16from sparkle.CLI.help import snapshot_help as snh
17from sparkle.CLI.help import jobs as jobs_help
18from sparkle.CLI.help import resolve_instance_name
21def parser_function() -> argparse.ArgumentParser:
22 """Define the command line arguments."""
23 parser = argparse.ArgumentParser(
24 description="Command to clean files from the platform."
25 )
26 parser.add_argument(*ac.CleanupArgumentAll.names, **ac.CleanupArgumentAll.kwargs)
27 parser.add_argument(*ac.CleanupArgumentLogs.names, **ac.CleanupArgumentLogs.kwargs)
28 parser.add_argument(
29 *ac.CleanupArgumentRemove.names, **ac.CleanupArgumentRemove.kwargs
30 )
31 parser.add_argument(
32 *ac.CleanUpPerformanceDataArgument.names,
33 **ac.CleanUpPerformanceDataArgument.kwargs,
34 )
35 parser.add_argument(
36 *ac.CleanUpFeatureDataArgument.names,
37 **ac.CleanUpFeatureDataArgument.kwargs,
38 )
39 return parser
42def check_logs_performance_data(performance_data: PerformanceDataFrame) -> int:
43 """Check if the performance data is missing values that can be extracted from the logs.
45 Args:
46 performance_data (PerformanceDataFrame): The performance data.
48 Returns:
49 int: The number of updated values.
50 """
51 # empty_indices = performance_data.empty_indices
52 pattern = re.compile(
53 r"^(?P<objective>\S+)\s*,\s*"
54 r"(?P<instance_set>\S+)\s*,\s*"
55 r"(?P<instance>\S+)\s*,\s*"
56 r"(?P<run_id>\S+)\s*\|\s*"
57 r"(?P<solver>\S+)\s*,\s*"
58 r"(?P<config_id>\S+)\s*:\s*"
59 r"(?P<target_value>\S+)$"
60 )
62 # Only iterate over slurm log files
63 log_files = [
64 file
65 for file in gv.settings().DEFAULT_log_output.glob("**/*")
66 if file.is_file() and file.suffix == ".out"
67 ]
68 count = 0
69 for log in log_files:
70 for line in log.read_text().splitlines():
71 match = pattern.match(line)
72 if match:
73 objective = match.group("objective")
74 instance = match.group("instance")
75 run_id = int(match.group("run_id"))
76 solver = match.group("solver")
77 config_id = match.group("config_id")
78 target_value = match.group("target_value")
79 # The log records the (set_name, instance_name) pair directly.
80 instance_pair = (match.group("instance_set"), instance)
81 if instance_pair not in performance_data.instance_pairs:
82 continue # Unknown instance, skip
83 current_value = performance_data.get_value(
84 solver, instance_pair, config_id, objective, run_id
85 )
86 # TODO: Would be better to extract all nan indices from PDF and check against this?
87 if (
88 (
89 isinstance(current_value, (int, float))
90 and math.isnan(current_value)
91 )
92 or isinstance(current_value, str)
93 and current_value == "nan"
94 ):
95 performance_data.set_value(
96 target_value, solver, instance_pair, config_id, objective, run_id
97 )
98 count += 1
99 if count:
100 performance_data.save_csv()
101 return count
104def check_logs_feature_data(feature_data: FeatureDataFrame) -> int:
105 """Check if the feature data is missing values that can be extracted from the logs.
107 Args:
108 feature_data (FeatureDataFrame): The feature data.
110 Returns:
111 int: The number of updated values.
112 """
113 # empty_indices = performance_data.empty_indices
114 pattern = re.compile(
115 r"^(?P<extractor>\S+)\s*"
116 r"(?P<instance_set>\S+)\s*"
117 r"(?P<instance>\S+)\s*"
118 r"(?P<feature_group>\S+)\s*"
119 r"(?P<feature_name>\S+)\s*\|\s*"
120 r"(?P<target_value>\S+)$"
121 )
123 # Only iterate over slurm log files
124 log_files = [
125 file
126 for file in gv.settings().DEFAULT_log_output.glob("**/*")
127 if file.is_file() and file.suffix == ".out"
128 ]
129 count = 0
130 for log in log_files:
131 for line in log.read_text().splitlines():
132 match = pattern.match(line)
133 if match:
134 target_value = float(match.group("target_value")) # Must be a float
135 if math.isnan(target_value):
136 continue
137 extractor = match.group("extractor")
138 instance = match.group("instance")
139 feature_group = match.group("feature_group")
140 feature_name = match.group("feature_name")
141 # The log records the (set_name, instance_name) pair directly.
142 instance_set = match.group("instance_set")
143 if (instance_set, instance) not in feature_data.instance_pairs:
144 continue # Unknown instance, skip
145 current_value = feature_data.get_value(
146 instance_set, instance, extractor, feature_group, feature_name
147 )
148 if (
149 (
150 isinstance(current_value, (int, float))
151 and math.isnan(current_value)
152 )
153 or isinstance(current_value, str)
154 and current_value == "nan"
155 ):
156 feature_data.set_value(
157 instance_set,
158 instance,
159 extractor,
160 feature_group,
161 feature_name,
162 target_value,
163 )
164 count += 1
165 if count:
166 feature_data.save_csv()
167 return count
170def remove_temporary_files() -> None:
171 """Remove temporary files. Only removes files not affecting the sparkle state."""
172 shutil.rmtree(gv.settings().DEFAULT_log_output, ignore_errors=True)
173 gv.settings().DEFAULT_log_output.mkdir()
176def main(argv: list[str]) -> None:
177 """Main function of the cleanup command."""
178 # Log command call
179 sl.log_command(sys.argv, gv.settings().random_state)
181 # Define command line arguments
182 parser = parser_function()
184 # Process command line arguments
185 args = parser.parse_args(argv)
187 if args.performance_data:
188 jobs_help.check_running_waiting_jobs(
189 gv.settings().DEFAULT_log_output, {DataFileLock.PERFORMANCE}
190 )
191 performance_data = PerformanceDataFrame(
192 gv.settings().DEFAULT_performance_data_path
193 )
194 count = check_logs_performance_data(performance_data)
195 print(
196 f"Extracted {count} values from the logs and placed them in the PerformanceDataFrame."
197 )
199 # Remove empty configurations
200 removed_configurations = 0
201 for solver, configurations in performance_data.configurations.items():
202 for config_id, config in configurations.items():
203 if config_id == PerformanceDataFrame.default_configuration:
204 continue
205 if not config: # Empty configuration, remove
206 performance_data.remove_configuration(solver, config_id)
207 removed_configurations += 1
208 if removed_configurations:
209 print(
210 f"Removed {removed_configurations} empty configurations from the "
211 "Performance DataFrame."
212 )
214 index_num = len(performance_data.index)
215 # We only clean lines that are completely empty
216 performance_data.remove_empty_runs()
217 print(
218 f"Removed {index_num - len(performance_data.index)} rows from the "
219 f"Performance DataFrame, leaving {len(performance_data.index)} rows."
220 )
222 # Sanity check all indices, clean lines that are broken
223 # NOTE: This check is quite e
224 objective_errors, instance_errors, run_id_errors = 0, 0, 0
225 known_objectives = [o.name for o in gv.settings().objectives]
226 wrong_indices = []
227 for objective, instance_set, instance, run_id in performance_data.index:
228 if objective not in known_objectives:
229 objective_errors += 1
230 wrong_indices.append((objective, instance_set, instance, run_id))
231 # print("Objective issue:", objective)
232 elif isinstance(run_id, str) and not run_id.isdigit():
233 run_id_errors += 1
234 wrong_indices.append((objective, instance_set, instance, run_id))
235 # print("Run id issue:", run_id)
236 else:
237 # NOTE: This check is very expensive, and it would be better if we could pass all the instances at once instead
238 instance_path = resolve_instance_name(
239 instance_set,
240 instance,
241 search_location=gv.settings().DEFAULT_instance_dir,
242 )
243 if instance_path is None:
244 instance_errors += 1
245 wrong_indices.append((objective, instance_set, instance, run_id))
246 if wrong_indices:
247 print(
248 f"Found {len(wrong_indices)} wrong indices in the PerformanceDataFrame ({objective_errors} objective errors, {instance_errors} instance errors, {run_id_errors} run id errors).\n"
249 "Removing from PerformanceDataFrame..."
250 )
251 performance_data.drop(wrong_indices, inplace=True)
252 print(
253 f"Removed {len(wrong_indices)} rows from the PerformanceDataFrame, leaving {len(performance_data.index)} rows."
254 )
255 performance_data.save_csv()
257 if args.feature_data:
258 jobs_help.check_running_waiting_jobs(
259 gv.settings().DEFAULT_log_output, {DataFileLock.FEATURE}
260 )
261 feature_data = FeatureDataFrame(gv.settings().DEFAULT_feature_data_path)
262 count = check_logs_feature_data(feature_data)
263 print(
264 f"Extracted {count} values from the logs and placed them in the FeatureDataFrame."
265 )
266 feature_data.save_csv()
267 # TODO: Can do other cleanup like index verification and empty line removal etc
268 # For example, we can check if each index references a valid instance, if not, remove the line
270 if args.all:
271 shutil.rmtree(gv.settings().DEFAULT_output, ignore_errors=True)
272 snh.create_working_dirs()
273 print("Removed all output files from the platform!")
274 elif args.remove:
275 snh.remove_current_platform()
276 snh.create_working_dirs()
277 print("Cleaned platform of all files!")
278 elif args.logs:
279 remove_temporary_files()
280 print("Cleaned platform of log files!")
281 elif not args.performance_data and not args.feature_data:
282 print(parser.print_help())
283 sys.exit(1)
284 sys.exit(0)
287if __name__ == "__main__":
288 main(sys.argv[1:])