Source code for moptipy.evaluation.ioh_analyzer

"""
Convert `moptipy` data to IOHanalyzer data.

The IOHanalyzer (https://iohanalyzer.liacs.nl/) is a tool that can analyze
the performance of iterative optimization heuristics in a wide variety of
ways. It is available both for local installation as well as online for
direct and free use (see, again, https://iohanalyzer.liacs.nl/). The
IOHanalyzer supports many of the diagrams that our evaluation utilities
provide - and several more. Here we provide the function
:func:`moptipy_to_ioh_analyzer` which converts the data generated by the
`moptipy` experimentation function
:func:`~moptipy.api.experiment.run_experiment` to the format that the
IOHanalyzer understands, as documented at
https://iohprofiler.github.io/IOHanalyzer/data/.

Notice that we here have implemented the meta data format version
"0.3.2 and below", as described at https://iohprofiler.github.io/IOHanalyzer\
/data/#iohexperimenter-version-032-and-below.

1. Carola Doerr, Furong Ye, Naama Horesh, Hao Wang, Ofer M. Shir, and Thomas
   Bäck. Benchmarking Discrete Optimization Heuristics with IOHprofiler.
   *Applied Soft Computing* 88(106027):1-21. March 2020.
   doi: https://doi.org/10.1016/j.asoc.2019.106027},
2. Carola Doerr, Hao Wang, Furong Ye, Sander van Rijn, and Thomas Bäck.
   *IOHprofiler: A Benchmarking and Profiling Tool for Iterative Optimization
   Heuristics.* October 15, 2018. New York, NY, USA: Cornell University,
   Cornell Tech. arXiv:1810.05281v1 [cs.NE] 11 Oct 2018.
   https://arxiv.org/pdf/1810.05281.pdf
3. Hao Wang, Diederick Vermetten, Furong Ye, Carola Doerr, and Thomas Bäck.
   IOHanalyzer: Detailed Performance Analyses for Iterative Optimization
   Heuristics. *ACM Transactions on Evolutionary Learning and Optimization*
   2(1)[3]:1-29. March 2022.doi: https://doi.org/10.1145/3510426.
4. Jacob de Nobel and Furong Ye and Diederick Vermetten and Hao Wang and
   Carola Doerr and Thomas Bäck. *IOHexperimenter: Benchmarking Platform for
   Iterative Optimization Heuristics.* November 2021. New York, NY, USA:
   Cornell University, Cornell Tech. arXiv:2111.04077v2 [cs.NE] 17 Apr 2022.
   https://arxiv.org/pdf/2111.04077.pdf
5. Data Format: Iterative Optimization Heuristics Profiler.
   https://iohprofiler.github.io/IOHanalyzer/data/
"""

import argparse
import contextlib
from typing import Any, Callable, Final

import numpy as np
from pycommons.io.console import logger
from pycommons.io.path import Path, directory_path
from pycommons.strings.string_conv import float_to_str
from pycommons.types import check_int_range, type_error

from moptipy.evaluation.base import F_NAME_RAW, TIME_UNIT_FES, check_f_name
from moptipy.evaluation.progress import Progress
from moptipy.utils.help import moptipy_argparser


def __prefix(s: str) -> str:
    """
    Return `xxx` if `s` is of the form `xxx_i` and `i` is `int`.

    :param s: the function name
    :return: the dimension
    """
    idx = s.rfind("_")
    if idx > 0:
        with contextlib.suppress(ValueError):
            i = int(s[idx + 1:])
            if i > 0:
                return s[:idx].strip()
    return s


def __int_suffix(s: str) -> int:
    """
    Return `i` if `s` is of the form `xxx_i` and `i` is `int`.

    This function tries to check if the name

    :param s: the function name
    :return: the dimension
    """
    idx = s.rfind("_")
    if idx > 0:
        with contextlib.suppress(ValueError):
            i = int(s[idx + 1:])
            if i > 0:
                return i
    return 1


def __npstr(a: Any) -> str:
    """
    Convert numpy numbers to strings.

    :param a: the input
    :returns: a string
    """
    return str(int(a)) if isinstance(a, np.integer) \
        else float_to_str(float(a))


[docs] def moptipy_to_ioh_analyzer( results_dir: str, dest_dir: str, inst_name_to_func_id: Callable[[str], str] = __prefix, inst_name_to_dimension: Callable[[str], int] = __int_suffix, inst_name_to_inst_id: Callable[[str], int] = lambda x: 1, suite: str = "moptipy", f_name: str = F_NAME_RAW, f_standard: dict[str, int | float] | None = None) -> None: """ Convert moptipy log data to IOHanalyzer log data. :param results_dir: the directory where we can find the results in moptipy format :param dest_dir: the directory where we would write the IOHanalyzer style data :param inst_name_to_func_id: convert the instance name to a function ID :param inst_name_to_dimension: convert an instance name to a function dimension :param inst_name_to_inst_id: convert the instance name an instance ID, which must be a positive integer number :param suite: the suite name :param f_name: the objective name :param f_standard: a dictionary mapping instances to standard values """ source: Final[Path] = directory_path(results_dir) dest: Final[Path] = Path(dest_dir) dest.ensure_dir_exists() logger(f"converting the moptipy log files in {source!r} to " f"IOHprofiler data in {dest!r}. First we load the data.") if (f_standard is not None) and (not isinstance(f_standard, dict)): raise type_error(f_standard, "f_standard", dict) if not isinstance(suite, str): raise type_error(suite, "suite", str) if (len(suite) <= 0) or (" " in suite): raise ValueError(f"invalid suite name {suite!r}") if not callable(inst_name_to_func_id): raise type_error( inst_name_to_func_id, "inst_name_to_func_id", call=True) if not callable(inst_name_to_dimension): raise type_error( inst_name_to_dimension, "inst_name_to_dimension", call=True) if not callable(inst_name_to_inst_id): raise type_error( inst_name_to_inst_id, "inst_name_to_inst_id", call=True) # the data data: Final[dict[str, dict[str, dict[int, list[ tuple[int, np.ndarray, np.ndarray]]]]]] = {} # this consumer collects all the data in a structured fashion def __consume(progress: Progress) -> None: nonlocal data # noqa nonlocal inst_name_to_func_id nonlocal inst_name_to_dimension nonlocal inst_name_to_inst_id _algo: dict[str, dict[int, list[tuple[int, np.ndarray, np.ndarray]]]] if progress.algorithm in data: _algo = data[progress.algorithm] else: data[progress.algorithm] = _algo = {} _func_id: Final[str] = inst_name_to_func_id(progress.instance) if not isinstance(_func_id, str): raise type_error(_func_id, "function id", str) if (len(_func_id) <= 0) or ("_" in _func_id): raise ValueError(f"invalid function id {_func_id!r}.") _func: dict[int, list[tuple[int, np.ndarray, np.ndarray]]] if _func_id in _algo: _func = _algo[_func_id] else: _algo[_func_id] = _func = {} _dim: Final[int] = check_int_range( inst_name_to_dimension(progress.instance), "dimension", 1) _iid: Final[int] = check_int_range( inst_name_to_inst_id(progress.instance), "instance id", 1) _res: Final[tuple[int, np.ndarray, np.ndarray]] = \ (_iid, progress.time, progress.f) if _dim in _func: _func[_dim].append(_res) else: _func[_dim] = [_res] Progress.from_logs(source, consumer=__consume, time_unit=TIME_UNIT_FES, f_name=check_f_name(f_name), f_standard=f_standard, only_improvements=True) if len(data) <= 0: raise ValueError("did not find any data!") logger(f"finished loading data from {len(data)} algorithms, " "now writing output.") for algo_name in sorted(data.keys()): algo = data[algo_name] algo_dir: Path = dest.resolve_inside(algo_name) algo_dir.ensure_dir_exists() logger(f"writing output for {len(algo)} functions of " f"algorithm {algo_name!r}.") for func_id in sorted(algo.keys()): func_dir: Path = algo_dir.resolve_inside(f"data_f{func_id}") func_dir.ensure_dir_exists() func = algo[func_id] logger(f"writing output for algorithm {algo_name!r} and " f"function {func_id!r}, got {len(func)} dimensions.") func_name = f"IOHprofiler_f{func_id}" with algo_dir.resolve_inside( f"{func_name}.info").open_for_write() as info: for dimi in sorted(func.keys()): dim_path = func_dir.resolve_inside( f"{func_name}_DIM{dimi}.dat") info.write(f"suite = {suite!r}, funcId = {func_id!r}, " f"DIM = {dimi}, algId = {algo_name!r}\n") info.write("%\n") info.write(dim_path[len(algo_dir) + 1:]) with dim_path.open_for_write() as dat: for per_dim in sorted( func[dimi], key=lambda x: (x[0], x[2][-1], x[1][-1])): info.write(f", {per_dim[0]}:") fes = per_dim[1] f = per_dim[2] info.write(__npstr(fes[-1])) info.write("|") info.write(__npstr(f[-1])) dat.write( '"function evaluation" "best-so-far f(x)"\n') for i, ff in enumerate(f): dat.write( f"{__npstr(fes[i])} {__npstr(ff)}\n") dat.write("\n") info.write("\n") del data logger("finished converting moptipy data to IOHprofiler data.")
# Run conversion if executed as script if __name__ == "__main__": parser: Final[argparse.ArgumentParser] = moptipy_argparser( __file__, "Convert experimental results from the moptipy to the " "IOHanalyzer format.", "The experiment execution API of moptipy creates an output " "folder structure with clearly specified log files that can be" " evaluated with our experimental data analysis API. The " "IOHprofiler tool chain offers another format (specified in " "https://iohprofiler.github.io/IOHanalyzer/data/). With this " "tool here, you can convert from the moptipy to the " "IOHprofiler format.") parser.add_argument( "source", help="the directory with moptipy log files", type=Path, nargs="?", default="./results") parser.add_argument( "dest", help="the directory to write the IOHanalyzer data to", type=Path, nargs="?", default="./IOHanalyzer") args: Final[argparse.Namespace] = parser.parse_args() moptipy_to_ioh_analyzer(args.source, args.dest)