Coverage for moptipy/evaluation/mo_end_results.py: 79%
193 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-03 07:20 +0000
1"""A set of end results from a multi-objective run."""
3import argparse
4from dataclasses import dataclass
5from itertools import chain
6from typing import Any, Callable, Final, Generator, Iterable, cast
8from pycommons.ds.sequences import reiterable
9from pycommons.io.console import logger
10from pycommons.io.csv import (
11 CSV_SEPARATOR,
12 csv_column,
13 csv_scope,
14 csv_val_or_none,
15)
16from pycommons.io.parser import Parser
17from pycommons.io.path import Path, file_path, write_lines
18from pycommons.strings.string_conv import (
19 num_to_str,
20 str_to_num,
21)
22from pycommons.types import type_error
24from moptipy.api.logging import (
25 SECTION_ARCHIVE_QUALITY,
26 SECTION_PROGRESS,
27)
28from moptipy.evaluation.end_results import CsvReader as CsvReaderBase
29from moptipy.evaluation.end_results import CsvWriter as CsvWriterBase
30from moptipy.evaluation.end_results import EndResult
31from moptipy.evaluation.end_results import EndResultLogParser as _Erlp
32from moptipy.utils.help import moptipy_argparser
33from moptipy.utils.logger import (
34 SECTION_END,
35 SECTION_START,
36)
37from moptipy.utils.math import try_int
40@dataclass(frozen=True, init=False, order=False, eq=False)
41class MOEndResult(EndResult):
42 """A multi-objective end result record."""
44 #: The objective values for the subordinate objective functions.
45 fs: tuple[int | float, ...]
47 def __init__(self,
48 algorithm: str,
49 instance: str,
50 objective: str,
51 encoding: str | None,
52 rand_seed: int,
53 best_f: int | float,
54 last_improvement_fe: int,
55 last_improvement_time_millis: int,
56 total_fes: int,
57 total_time_millis: int,
58 goal_f: int | float | None,
59 max_fes: int | None,
60 max_time_millis: int | None,
61 fs: tuple[int | float, ...]) -> None:
62 """
63 Create the multi-objective end result record.
65 :param algorithm: the algorithm name
66 :param instance: the instance name
67 :param objective: the name of the objective function
68 :param encoding: the name of the encoding that was used, if any, or
69 `None` if no encoding was used
70 :param rand_seed: the random seed
71 :param best_f: the best reached objective value
72 :param last_improvement_fe: the FE when best_f was reached
73 :param last_improvement_time_millis: the time when best_f was reached
74 :param total_fes: the total FEs
75 :param total_time_millis: the total runtime
76 :param goal_f: the goal objective value, if provide
77 :param max_fes: the optional maximum FEs
78 :param max_time_millis: the optional maximum runtime
79 :param fs: the objective value vectors
81 :raises TypeError: if any parameter has a wrong type
82 :raises ValueError: if the parameter values are inconsistent
83 """
84 super().__init__(
85 algorithm=algorithm,
86 instance=instance,
87 objective=objective,
88 encoding=encoding,
89 rand_seed=rand_seed,
90 best_f=best_f,
91 last_improvement_fe=last_improvement_fe,
92 last_improvement_time_millis=last_improvement_time_millis,
93 total_fes=total_fes,
94 total_time_millis=total_time_millis,
95 goal_f=goal_f,
96 max_fes=max_fes,
97 max_time_millis=max_time_millis)
98 fsc = tuple.__len__(fs)
99 if fsc <= 0:
100 raise ValueError("Number of objectives must be greater than 0.")
101 fsu: list[int | float] = []
102 changed: bool = False
103 for val in fs:
104 val2 = try_int(val)
105 if val2 is not val:
106 changed = True
107 fsu.append(val2)
108 object.__setattr__(self, "fs", tuple(fsu) if changed else fs)
110 def _tuple(self) -> tuple[Any, ...]:
111 """
112 Get the comparison tuple.
114 :return: the comparison tuple
115 """
116 cr = list(super()._tuple())
117 cr.extend(self.fs)
118 return tuple(cr)
121def to_csv(results: Iterable[EndResult], file: str) -> Path:
122 """
123 Write a sequence of end results to a file in CSV format.
125 :param results: the end results
126 :param file: the path
127 :return: the path of the file that was written
128 """
129 path: Final[Path] = Path(file)
130 logger(f"Writing end results to CSV file {path!r}.")
131 path.ensure_parent_dir_exists()
132 with path.open_for_write() as wt:
133 write_lines(CsvWriter.write(results), wt)
134 logger(f"Done writing end results to CSV file {path!r}.")
135 return path
138def from_csv(file: str,
139 filterer: Callable[[EndResult], bool]
140 = lambda _: True) -> Generator[
141 EndResult | MOEndResult, None, None]:
142 """
143 Parse a given CSV file to get :class:`MOEndResult` Records.
145 :param file: the path to parse
146 :param filterer: an optional filter function
147 """
148 path: Final[Path] = file_path(file)
149 logger(f"Now reading CSV file {path!r}.")
150 with path.open_for_read() as rd:
151 for r in CsvReader.read(rd):
152 if filterer(r):
153 yield r
154 logger(f"Done reading CSV file {path!r}.")
157def from_logs(path: str) -> Generator[EndResult | MOEndResult, None, None]:
158 """
159 Parse a given path and yield all (multi-objective) end results found.
161 If `path` identifies a file with suffix `.txt`, then this file is
162 parsed. The appropriate :class:`moptipy.evaluation.end_results.EndResult`
163 or :class:`MOEndResult` is created and yielded.
164 If `path` identifies a directory, then this directory is parsed
165 recursively for each log file found, one record is yielded.
167 :param path: the path to parse
168 """
169 for group in __MOEndResultLogParser().parse(path):
170 yield from group
173class CsvWriter(CsvWriterBase):
174 """A class for CSV writing of `EndResult` records."""
176 def __init__(self, data: Iterable[EndResult],
177 scope: str | None = None) -> None:
178 """
179 Initialize the csv writer.
181 :param data: the data
182 :param scope: the prefix to be pre-pended to all columns
183 """
184 data = reiterable(data)
185 super().__init__(data, scope)
187 #: do we need the encoding?
188 self.__fcols: Final[int] = max(
189 tuple.__len__(er.fs) if isinstance(
190 er, MOEndResult) else 0 for er in data)
192 def get_column_titles(self) -> Iterable[str]:
193 """
194 Get the column titles.
196 :returns: the column titles
197 """
198 p: Final[str | None] = self.scope
199 return chain(super().get_column_titles(), (
200 csv_scope(p, x) for x in (
201 f"f{i}" for i in range(self.__fcols))))
203 def get_row(self, data: EndResult) -> Iterable[str]:
204 """
205 Render a single end result record to a CSV row.
207 :param data: the end result record
208 :returns: the row iterator
209 """
210 yield from super().get_row(data)
211 if isinstance(data, MOEndResult):
212 yield from map(num_to_str, data.fs)
214 def get_header_comments(self) -> Iterable[str]:
215 """
216 Get any possible header comments.
218 :returns: the header comments
219 """
220 return ("Multi-Objective Experiment End Results",
221 "See the description at the bottom of the file.")
223 def get_footer_comments(self) -> Iterable[str]:
224 """
225 Get any possible footer comments.
227 :returns: the footer comments
228 """
229 yield from super().get_footer_comments()
230 for i in range(self.__fcols):
231 yield (f"f{i}: the objective value computed with "
232 f"the {i + 1}-th objective function.")
235class CsvReader(CsvReaderBase):
236 """A csv parser for end results."""
238 def __init__(self, columns: dict[str, int]) -> None:
239 """
240 Create a CSV parser for `EndResult` records.
242 :param columns: the columns
243 """
244 super().__init__(columns)
245 i: int = 0
246 fcols: Final[list[int]] = []
247 while True:
248 colname: str = f"f{i}"
249 try:
250 fcols.append(csv_column(columns, colname))
251 except KeyError:
252 break
253 i += 1
254 #: the objective value columns
255 self.__fcols: Final[tuple[int, ...]] = tuple(fcols)
257 def parse_row(self, data: list[str]) -> EndResult | MOEndResult:
258 """
259 Parse a row of data.
261 :param data: the data row
262 :return: the end result statistics
263 """
264 res = super().parse_row(data)
265 vals: Final[list[int | float]] = []
266 for col in self.__fcols:
267 v = csv_val_or_none(data, col, str_to_num)
268 if v is None:
269 break
270 vals.append(v)
271 if list.__len__(vals) <= 0:
272 return res
273 return MOEndResult(
274 algorithm=res.algorithm,
275 instance=res.instance,
276 objective=res.objective,
277 encoding=res.encoding,
278 rand_seed=res.rand_seed,
279 best_f=res.best_f,
280 last_improvement_fe=res.last_improvement_fe,
281 last_improvement_time_millis=res.last_improvement_time_millis,
282 total_fes=res.total_fes,
283 total_time_millis=res.total_time_millis,
284 goal_f=res.goal_f,
285 max_fes=res.max_fes,
286 max_time_millis=res.max_time_millis,
287 fs=tuple(vals))
290class __MOEndResultLogParser(Parser[Iterable[EndResult]]):
291 """The internal log parser class."""
293 def _parse_file(self, file: Path) -> Iterable[EndResult]:
294 """
295 Get the parsing result.
297 :returns: the `EndResult` instance
298 """
299 self._progress_logger(
300 f"Beginning multi-objective parsing of file {file!r}.")
301 o: Final[EndResult] = _Erlp().parse_file(file)
302 if not isinstance(o, EndResult):
303 raise type_error(o, f"parse({file!r})", EndResult)
305 with file.open_for_read() as reader:
306 lines: tuple[str, ...] = tuple(map(str.strip, str.splitlines(
307 reader.read())))
308 count = tuple.__len__(lines)
309 if count <= 2:
310 raise ValueError(
311 f"Inconsistent number {count} of lines in file {file!r}")
313 # process the archive
314 archive: Final[list[tuple[int | float, ...]]] = []
315 begin: str = f"{SECTION_START}{SECTION_ARCHIVE_QUALITY}"
316 end: str = f"{SECTION_END}{SECTION_ARCHIVE_QUALITY}"
317 state: int = 0
318 for line in lines:
319 if line == begin:
320 if state != 0:
321 raise ValueError(f"Inconsistent begin state in "
322 f"file {file!r} vs. {begin!r}/{end!r}.")
323 state = 1
324 continue
325 if line == end:
326 if state != 2:
327 raise ValueError(f"Inconsistent end state in "
328 f"file {file!r} vs. {begin!r}/{end!r}.")
329 state = 3
330 break
331 if state == 1:
332 if line.startswith("f"):
333 state = 2
334 continue
335 state = 2
336 if state != 2:
337 continue
338 try:
339 archive.append(tuple(map(str_to_num, map(str.strip, str.split(
340 line, CSV_SEPARATOR)))))
341 except ValueError as ve:
342 raise ValueError(
343 f"Error when parsing line {line!r} of file {file!r} in "
344 f"{SECTION_ARCHIVE_QUALITY}.") from ve
345 if state != 3:
346 return (o, )
348 # first, we find the progress
349 progress: Final[list[tuple[int | float, ...]]] = []
350 begin = f"{SECTION_START}{SECTION_PROGRESS}"
351 end = f"{SECTION_END}{SECTION_PROGRESS}"
352 state = 0
353 for line in lines:
354 if line == begin:
355 if state != 0:
356 raise ValueError(f"Inconsistent begin state in "
357 f"file {file!r} vs. {begin!r}/{end!r}.")
358 state = 1
359 continue
360 if line == end:
361 if state != 2:
362 raise ValueError(f"Inconsistent end state in "
363 f"file {file!r} vs. {begin!r}/{end!r}.")
364 state = 3
365 break
366 if state == 1:
367 if line.startswith("fes"):
368 state = 2
369 continue
370 state = 2
371 if state != 2:
372 continue
373 try:
374 progress.append(tuple(map(str_to_num, map(str.strip, str.split(
375 line, CSV_SEPARATOR)))))
376 except ValueError as ve:
377 raise ValueError(
378 f"Error when parsing line {line!r} of file {file!r} "
379 f"in {SECTION_PROGRESS}.") from ve
380 if state not in {0, 3}:
381 raise ValueError(f"Inconsistent state {state} in "
382 f"file {file!r} vs. {begin!r}/{end!r}.")
384 count = list.__len__(archive)
385 if count < 1:
386 raise ValueError(f"No solution archived in file {file!r}.")
388 dim: Final[int] = tuple.__len__(archive[0])
389 for solution in archive:
390 if tuple.__len__(solution) != dim:
391 raise ValueError(
392 f"Inconsistent archive dimension of {solution} in file "
393 f"{file!r}, should be {dim}.")
395 for time in progress:
396 if tuple.__len__(time) != dim + 2:
397 raise ValueError(
398 "Inconsistent progress dimension of record "
399 f"{time} in {file!r}, should be {dim + 2}.")
401 out: list[MOEndResult] = []
402 for solution in archive:
403 found: tuple[int | float, ...] | None = None
404 for time in progress:
405 if time[-dim:] == solution:
406 found = time
407 break
408 out.append(MOEndResult(
409 algorithm=o.algorithm,
410 instance=o.instance,
411 objective=o.objective,
412 encoding=o.encoding,
413 rand_seed=o.rand_seed,
414 best_f=solution[0],
415 last_improvement_fe=o.last_improvement_fe
416 if found is None else cast("int", found[0]),
417 last_improvement_time_millis=o.last_improvement_time_millis
418 if found is None else cast("int", found[1]),
419 total_fes=o.total_fes,
420 total_time_millis=o.total_time_millis,
421 goal_f=o.goal_f,
422 max_fes=o.max_fes,
423 max_time_millis=o.max_time_millis,
424 fs=solution[1:]))
425 self._progress_logger(f"Done parsing file {file!r} multi-objectively.")
426 return out
429# Run log files to end results if executed as script
430if __name__ == "__main__":
431 parser: Final[argparse.ArgumentParser] = moptipy_argparser(
432 __file__,
433 "Convert multi-objective log files obtained with moptipy to the "
434 "end results CSV format that can be post-processed or exported to "
435 "other tools.",
436 "This program recursively parses a folder hierarchy created by"
437 " the moptipy multi-objective experiment execution facility. "
438 "This folder structure follows the scheme of algorithm/instance/"
439 "log_file and has one log file per run. As result of the parsing, "
440 "one CSV file (where columns are separated by ';') is created with"
441 " one row per log file. This row contains the end-of-run state"
442 " loaded from the log file. Whereas the log files may store "
443 "the complete progress of one run of one algorithm on one "
444 "problem instance as well as the algorithm configuration "
445 "parameters, instance features, system settings, and the final"
446 " results, the end results CSV file will only represent the "
447 "final result quality, when it was obtained, how long the runs"
448 " took, etc. This information is much denser and smaller and "
449 "suitable for importing into other tools such as Excel or for "
450 "postprocessing.")
451 parser.add_argument(
452 "source", nargs="?", default="./results",
453 help="the location of the experimental results, i.e., the root folder "
454 "under which to search for log files", type=Path)
455 parser.add_argument(
456 "dest", help="the path to the end results CSV file to be created",
457 type=Path, nargs="?", default="./evaluation/end_results.txt")
458 args: Final[argparse.Namespace] = parser.parse_args()
460 to_csv(from_logs(args.source), args.dest)