significantly improved Generator-based API

Author: thomasWeise

README.md

@@ -1697,11 +1697,8 @@ with temp_dir() as td:
         max_fes=10000,  # we grant 10000 FEs per run
         n_runs=4)  # perform 4 runs per algorithm * instance combination
 
-    end_results = list(from_logs(td))  # get results from log files
-
-    end_stats = []  # the list to receive the statistics records
-    # compute end result statistics for all algorithm+instance combinations
-    from_end_results(end_results, end_stats.append)
+    # Compute the end statistics from end results loaded from log files.
+    end_stats = list(from_end_results(from_logs(td)))
 
     # store the statistics to a CSV file
     es_csv = to_csv(end_stats, td.resolve_inside("end_stats.txt"))

examples/end_statistics_jssp.py

@@ -23,11 +23,8 @@
         max_fes=10000,  # we grant 10000 FEs per run
         n_runs=4)  # perform 4 runs per algorithm * instance combination
 
-    end_results = list(from_logs(td))  # get results from log files
-
-    end_stats = []  # the list to receive the statistics records
-    # compute end result statistics for all algorithm+instance combinations
-    from_end_results(end_results, end_stats.append)
+    # Compute the end statistics from end results loaded from log files.
+    end_stats = list(from_end_results(from_logs(td)))
 
     # store the statistics to a CSV file
     es_csv = to_csv(end_stats, td.resolve_inside("end_stats.txt"))

examples/end_statistics_over_feature_plot.py

@@ -80,10 +80,8 @@ def make_rls(problem) -> Execution:
                    n_runs=7)  # we will execute 7 runs per setup
     # Once we arrived here, the 20*7 = 140 runs have completed.
 
-    end_results = list(from_logs(td))  # load results
-
-    end_stats = []  # the end statistics go into this list
-    from_end_results(end_results, end_stats.append)
+    # Compute the end statistics from end results loaded from log files.
+    end_stats = list(from_end_results(from_logs(td)))
 
     files = []  # the collection of files
 

examples/end_statistics_over_param_plot.py

@@ -37,6 +37,7 @@
 of `m` as `int(name[9:])` and the base name of the algorithm as `name[9:]`,
 i.e., `rls_flipB` in our example.
 """
+from itertools import chain
 from time import sleep
 from webbrowser import open_new_tab
 
@@ -95,11 +96,10 @@ def make_rls(problem, m: int) -> Execution:
 
     end_results = list(from_logs(td))  # load results
 
-    end_stats = []  # the end statistics go into this list
-    from_end_results(end_results, end_stats.append)
-    from_end_results(  # over all instances summary
-        end_results, end_stats.append, join_all_instances=True,
-        join_all_objectives=True)
+    end_stats = list(chain(  # Compute two sets of end statistics
+        from_end_results(end_results),  # One per algorithm-instance combo
+        from_end_results(  # and one on a per-algorithm basis.
+            end_results, join_all_instances=True, join_all_objectives=True)))
 
     files = []  # the collection of files
 

moptipy/evaluation/end_results.py

@@ -32,7 +32,7 @@
 )
 from pycommons.io.csv import CsvReader as CsvReaderBase
 from pycommons.io.csv import CsvWriter as CsvWriterBase
-from pycommons.io.path import Path, file_path, line_writer
+from pycommons.io.path import Path, file_path, write_lines
 from pycommons.strings.string_conv import (
     int_or_none_to_str,
     num_or_none_to_str,
@@ -459,9 +459,7 @@ def to_csv(results: Iterable[EndResult], file: str) -> Path:
     logger(f"Writing end results to CSV file {path!r}.")
     path.ensure_parent_dir_exists()
     with path.open_for_write() as wt:
-        consumer: Final[Callable[[str], None]] = line_writer(wt)
-        for r in CsvWriter.write(results):
-            consumer(r)
+        write_lines(CsvWriter.write(results), wt)
     logger(f"Done writing end results to CSV file {path!r}.")
     return path
 
@@ -722,12 +720,13 @@ def parse_row(self, data: list[str]) -> EndResult:
 class EndResultLogParser[T](SetupAndStateParser[T]):
     """The internal log parser class."""
 
-    def get_result(self) -> T:
+    def _parse_file(self, file: Path) -> T:
         """
         Get the parsing result.
 
         :returns: the :class:`EndResult` instance
         """
+        super()._parse_file(file)
         return cast(T, EndResult(self.algorithm,
                                  self.instance,
                                  self.objective,
@@ -794,16 +793,13 @@ def __init__(
         self.__hit_goal: bool = False
         self.__state: int = 0
 
-    def before_get_result(self) -> None:
-        """Before we get the result."""
+    def _parse_file(self, file: Path) -> EndResult:
+        super()._parse_file(file)
         if self.__state != 2:
             raise ValueError(
                 "Illegal state, log file must have a "
                 f"{SECTION_PROGRESS!r} section.")
         self.__state = 0
-        return super().before_get_result()
-
-    def get_result(self) -> EndResult:
         __hit_goal = self.__hit_goal
         stop_fes: int = self.__stop_fes
         stop_ms: int = self.__stop_ms
@@ -832,8 +828,12 @@ def get_result(self) -> EndResult:
             max_time_millis=_join_goals(
                 self.__limit_ms_n, self.max_time_millis, min))
 
-    def after_get_result(self) -> None:
-        """Cleanup."""
+    def _end_parse_file(self, file: Path) -> None:
+        """
+        Cleanup.
+
+        :param file: the file that was parsed.
+        """
         self.__stop_fes = None
         self.__stop_ms = None
         self.__stop_f = None
@@ -846,65 +846,63 @@ def after_get_result(self) -> None:
         self.__limit_f_n = None
         self.__limit_f = -inf
         self.__hit_goal = False
-        super().after_get_result()
-
-    def start_parse_file(self, root: Path, current: Path) -> bool:
-        if super().start_parse_file(root, current):
-            if (self.algorithm is None) or (self.instance is None):
-                raise ValueError(
-                    f"Invalid state: algorithm={self.algorithm!r}, "
-                    f"instance={self.instance!r}.")
-
-            fes = self.__src_limit_fes(self.algorithm, self.instance) \
-                if callable(self.__src_limit_fes) else self.__src_limit_fes
-            self.__limit_fes_n = None if fes is None else \
-                check_int_range(fes, "limit_fes", 1, 1_000_000_000_000_000)
-            self.__limit_fes = inf if self.__limit_fes_n is None \
-                else self.__limit_fes_n
-
-            time = self.__src_limit_ms(self.algorithm, self.instance) \
-                if callable(self.__src_limit_ms) else self.__src_limit_ms
-            self.__limit_ms_n = None if time is None else \
-                check_int_range(time, "__limit_ms", 1, 1_000_000_000_000)
-            self.__limit_ms = inf if self.__limit_ms_n is None \
-                else self.__limit_ms_n
-
-            self.__limit_f_n = self.__src_limit_f(
-                self.algorithm, self.instance) \
-                if callable(self.__src_limit_f) else self.__src_limit_f
-            if self.__limit_f_n is not None:
-                if not isinstance(self.__limit_f_n, int | float):
-                    raise type_error(self.__limit_f_n, "limit_f", (
-                        int, float))
-                if not isfinite(self.__limit_f_n):
-                    if self.__limit_f_n <= -inf:
-                        self.__limit_f_n = None
-                    else:
-                        raise ValueError(
-                            f"invalid limit f={self.__limit_f_n} for "
-                            f"{self.algorithm} on {self.instance}")
-            self.__limit_f = -inf if self.__limit_f_n is None \
-                else self.__limit_f_n
-            return True
-        return False
-
-    def start_section(self, title: str) -> bool:
+        super()._end_parse_file(file)
+
+    def _start_parse_file(self, file: Path) -> None:
+        super()._start_parse_file(file)
+        a: Final[str | None] = self.algorithm
+        i: Final[str | None] = self.instance
+
+        fes = self.__src_limit_fes(a, i) if (a and i and callable(
+            self.__src_limit_fes)) else (
+            self.__src_limit_fes if isinstance(self.__src_limit_fes, int)
+            else None)
+        self.__limit_fes_n = None if fes is None else \
+            check_int_range(fes, "limit_fes", 1, 1_000_000_000_000_000)
+        self.__limit_fes = inf if self.__limit_fes_n is None \
+            else self.__limit_fes_n
+
+        time = self.__src_limit_ms(a, i) if (a and i and callable(
+            self.__src_limit_ms)) else (
+            self.__src_limit_ms if isinstance(self.__src_limit_ms, int)
+            else None)
+        self.__limit_ms_n = None if time is None else \
+            check_int_range(time, "__limit_ms", 1, 1_000_000_000_000)
+        self.__limit_ms = inf if self.__limit_ms_n is None \
+            else self.__limit_ms_n
+
+        self.__limit_f_n = self.__src_limit_f(a, i) if (a and i and callable(
+            self.__src_limit_f)) else (
+            self.__src_limit_f if isinstance(self.__src_limit_f, int | float)
+            else None)
+        if self.__limit_f_n is not None:
+            if not isinstance(self.__limit_f_n, int | float):
+                raise type_error(self.__limit_f_n, "limit_f", (
+                    int, float))
+            if not isfinite(self.__limit_f_n):
+                if self.__limit_f_n <= -inf:
+                    self.__limit_f_n = None
+                else:
+                    raise ValueError(
+                        f"invalid limit f={self.__limit_f_n} for "
+                        f"{self.algorithm} on {self.instance}")
+        self.__limit_f = -inf if self.__limit_f_n is None \
+            else self.__limit_f_n
+
+    def _start_section(self, title: str) -> bool:
         if title == SECTION_PROGRESS:
             if self.__state != 0:
                 raise ValueError(f"Already did section {title}.")
             self.__state = 1
             return True
-        return super().start_section(title)
-
-    def needs_more_lines(self) -> bool:
-        return (self.__state < 2) or super().needs_more_lines()
+        return super()._start_section(title)
 
-    def lines(self, lines: list[str]) -> bool:
-        if not isinstance(lines, list):
-            raise type_error(lines, "lines", list)
+    def _needs_more_lines(self) -> bool:
+        return (self.__state < 2) or super()._needs_more_lines()
 
+    def _lines(self, lines: list[str]) -> bool:
         if self.__state != 1:
-            return super().lines(lines)
+            return super()._lines(lines)
         self.__state = 2
 
         n_rows = len(lines)
@@ -978,7 +976,7 @@ def lines(self, lines: list[str]) -> bool:
         self.__stop_f = stop_f
         self.__stop_li_fe = stop_li_fe
         self.__stop_li_ms = stop_li_ms
-        return self.needs_more_lines()
+        return self._needs_more_lines()
 
 
 def from_logs(

moptipy/evaluation/end_statistics.py

@@ -12,7 +12,7 @@
 import os.path
 from dataclasses import dataclass
 from math import ceil, inf, isfinite
-from typing import Any, Callable, Final, Iterable, Iterator, cast
+from typing import Callable, Final, Generator, Iterable, Iterator, cast
 
 from pycommons.ds.sequences import reiterable
 from pycommons.io.console import logger
@@ -28,7 +28,7 @@
 )
 from pycommons.io.csv import CsvReader as CsvReaderBase
 from pycommons.io.csv import CsvWriter as CsvWriterBase
-from pycommons.io.path import Path, file_path, line_writer
+from pycommons.io.path import Path, file_path, write_lines
 from pycommons.math.sample_statistics import (
     KEY_MEAN_ARITH,
     KEY_STDDEV,
@@ -767,17 +767,15 @@ def create(source: Iterable[EndResult]) -> EndStatistics:
 
 
 def from_end_results(source: Iterable[EndResult],
-                     consumer: Callable[[EndStatistics], Any],
                      join_all_algorithms: bool = False,
                      join_all_instances: bool = False,
                      join_all_objectives: bool = False,
-                     join_all_encodings: bool = False) -> None:
+                     join_all_encodings: bool = False) \
+        -> Generator[EndStatistics, None, None]:
     """
     Aggregate statistics over a stream of end results.
 
     :param source: the stream of end results
-    :param consumer: the destination to which the new records will be
-        sent, can be the `append` method of a :class:`list`
     :param join_all_algorithms: should the statistics be aggregated
         over all algorithms
     :param join_all_instances: should the statistics be aggregated
@@ -786,11 +784,10 @@ def from_end_results(source: Iterable[EndResult],
         all objectives?
     :param join_all_encodings: should statistics be aggregated over all
         encodings
+    :returns: iterates over the generated end statistics records
     """
     if not isinstance(source, Iterable):
         raise type_error(source, "source", Iterable)
-    if not callable(consumer):
-        raise type_error(consumer, "consumer", call=True)
     if not isinstance(join_all_algorithms, bool):
         raise type_error(join_all_algorithms,
                          "join_all_algorithms", bool)
@@ -803,7 +800,7 @@ def from_end_results(source: Iterable[EndResult],
 
     if (join_all_algorithms and join_all_instances
             and join_all_objectives and join_all_encodings):
-        consumer(create(source))
+        yield create(source)
         return
 
     sorter: dict[tuple[str, str, str, str], list[EndResult]] = {}
@@ -828,9 +825,9 @@ def from_end_results(source: Iterable[EndResult],
 
     if len(sorter) > 1:
         for key in sorted(sorter.keys()):
-            consumer(create(sorter[key]))
+            yield create(sorter[key])
     else:
-        consumer(create(next(iter(sorter.values()))))
+        yield create(next(iter(sorter.values())))  #: pylint: disable=R1708
 
 
 def to_csv(data: EndStatistics | Iterable[EndStatistics],
@@ -846,16 +843,14 @@ def to_csv(data: EndStatistics | Iterable[EndStatistics],
     logger(f"Writing end result statistics to CSV file {path!r}.")
     path.ensure_parent_dir_exists()
     with path.open_for_write() as wt:
-        consumer: Final[Callable[[str], None]] = line_writer(wt)
-        for r in CsvWriter.write(
-                (data, ) if isinstance(data, EndStatistics) else data):
-            consumer(r)
+        write_lines(CsvWriter.write(
+            (data, ) if isinstance(data, EndStatistics) else data), wt)
 
     logger(f"Done writing end result statistics to CSV file {path!r}.")
     return path
 
 
-def from_csv(file: str) -> Iterator[EndStatistics]:
+def from_csv(file: str) -> Generator[EndStatistics, None, None]:
     """
     Parse a CSV file and collect all encountered :class:`EndStatistics`.
 
@@ -1531,13 +1526,10 @@ def aggregate_over_parameter(
 
     stats: Final[list[EndStatistics]] = []
     for pv in sorted(param_map.keys()):
-        def __make(es: EndStatistics, ppv=pv, ss=stats.append) -> None:
-            ss(__PvEndStatistics(es, ppv))
-
-        from_end_results(
-            param_map[pv], cast(Callable[[EndStatistics], None], __make),
-            join_all_algorithms, join_all_instances, join_all_objectives,
-            join_all_encodings)
+        for ess in from_end_results(
+                param_map[pv], join_all_algorithms, join_all_instances,
+                join_all_objectives, join_all_encodings):
+            stats.append(__PvEndStatistics(ess, pv))
     return cast(Callable[[EndStatistics], int | float],
                 __PvEndStatistics.get_param_value), tuple(stats)
 
@@ -1590,19 +1582,18 @@ def __make(es: EndStatistics, ppv=pv, ss=stats.append) -> None:
     args: Final[argparse.Namespace] = parser.parse_args()
 
     src_path: Final[Path] = args.source
-    end_results: list[EndResult]
+    end_results: Iterable[EndResult]
     if src_path.is_file():
         logger(f"{src_path!r} identifies as file, load as end-results csv")
-        end_results = list(end_results_from_csv(src_path))
+        end_results = end_results_from_csv(src_path)
     else:
         logger(f"{src_path!r} identifies as directory, load it as log files")
-        end_results = list(end_results_from_logs(src_path))
+        end_results = end_results_from_logs(src_path)
 
     end_stats: Final[list[EndStatistics]] = []
-    from_end_results(
-        source=end_results, consumer=end_stats.append,
+    to_csv(from_end_results(
+        source=end_results,
         join_all_algorithms=args.join_algorithms,
         join_all_instances=args.join_instances,
         join_all_objectives=args.join_objectives,
-        join_all_encodings=args.join_encodings)
-    to_csv(end_stats, args.dest)
+        join_all_encodings=args.join_encodings), args.dest)