Source code for pypesto.optimize.load

"""Load/reconstitute results from history."""

import logging
import os
from pathlib import Path
from typing import Optional

import h5py
import numpy as np

import pypesto

from ..C import (
    FVAL,
    GRAD,
    HESS,
    HISTORY,
    RES,
    SRES,
    SUFFIXES_CSV,
    SUFFIXES_HDF5,
    TRACE,
    X,
)
from ..history import (
    CsvHistory,
    Hdf5History,
    HistoryOptions,
    HistoryTypeError,
    OptimizerHistory,
)
from ..problem import Problem
from ..result import OptimizeResult, OptimizerResult, Result
from .options import OptimizeOptions

logger = logging.getLogger(__name__)

EXITFLAG_LOADED_FROM_FILE = -99


[docs] def fill_result_from_history( result: OptimizerResult, optimizer_history: OptimizerHistory, optimize_options: OptimizeOptions = None, ) -> OptimizerResult: """Overwrite some values in the result object with values in the history. Parameters ---------- result: Result as reported from the used optimizer. optimizer_history: History of function values recorded by the objective. optimize_options: Options on e.g. how to override. Returns ------- result: The in-place modified result. """ if optimize_options is None: optimize_options = OptimizeOptions() # logging # function values history_fval, result_fval = optimizer_history.fval_min, result.fval fval_exist = history_fval is not None and result_fval is not None fval_match = fval_exist and np.isclose(history_fval, result_fval) if fval_exist and not fval_match: logger.debug( "Minimal function value mismatch: " f"history {history_fval:8e}, result {result_fval:8e}" ) # parameters history_x, result_x = optimizer_history.x_min, result.x x_exist = history_x is not None and result_x is not None x_match = x_exist and np.allclose(history_x, result_x) if x_exist and not x_match: logger.debug( f"Optimal parameter mismatch: history {history_x}, " f"result {result_x}" ) # counters # we only use our own counters here as optimizers may report differently for key in (FVAL, GRAD, HESS, RES, SRES): setattr( result, f"n_{key}", getattr(optimizer_history.history, f"n_{key}") ) # initial values result.x0 = optimizer_history.x0 result.fval0 = optimizer_history.fval0 # trace result.history = optimizer_history.history # if optimizer is trusted, don't overwrite/complement optimal point if not optimize_options.history_beats_optimizer: return result # exit flag and message if isinstance(optimizer_history.history, Hdf5History): if (message := optimizer_history.history.message) is not None: result.message = message if (exitflag := optimizer_history.history.exitflag) is not None: result.exitflag = exitflag # optimal point for key in (X, FVAL, GRAD, HESS, RES, SRES): hist_val = getattr(optimizer_history, f"{key}_min") # replace by history if history has entry, or point does not match # point recorded in result if hist_val is not None or not fval_match or not x_match: setattr(result, key, hist_val) return result
def read_history_from_file( problem: Optional[Problem], history_options: HistoryOptions, identifier: str, ) -> OptimizerHistory: """Read history from file. Parameters ---------- problem: The problem to find optimal parameters for. If ``None``, bounds will be assumed to be [-inf, inf] for checking for admissible points. identifier: Multistart id. history_options: Optimizer history options. """ if history_options.storage_file is None: raise ValueError("No history file specified.") # evaluate type suffix = Path(history_options.storage_file).suffix[1:] if suffix in SUFFIXES_CSV: history = CsvHistory( file=history_options.storage_file.format(id=identifier), options=history_options, load_from_file=True, ) elif suffix in SUFFIXES_HDF5: history = Hdf5History.load( id=identifier, file=history_options.storage_file.format(id=identifier), options=history_options, ) else: raise HistoryTypeError(suffix) x0 = history.get_x_trace(0) if problem: lb, ub = problem.lb, problem.ub else: lb = np.full_like(x0, fill_value=-np.inf) ub = np.full_like(x0, fill_value=np.inf) return OptimizerHistory( history=history, x0=x0, lb=lb, ub=ub, generate_from_history=True, )
[docs] def read_result_from_file( problem: Optional[Problem], history_options: HistoryOptions, identifier: str, ) -> OptimizerResult: """Fill an OptimizerResult from history. Parameters ---------- problem: The problem to find optimal parameters for. If ``None``, bounds will be assumed to be [-inf, inf] for checking for admissible points. identifier: Multistart id. history_options: Optimizer history options. """ opt_hist = read_history_from_file( problem=problem, history_options=history_options, identifier=identifier ) result = OptimizerResult( id=identifier, message='loaded from file', exitflag=EXITFLAG_LOADED_FROM_FILE, time=( max(opt_hist.history.get_time_trace()) if len(opt_hist.history) else 0.0 ), ) result.id = identifier result = fill_result_from_history( result=result, optimizer_history=opt_hist, ) if problem: result.update_to_full(problem) return result
[docs] def read_results_from_file( problem: Problem, history_options: HistoryOptions, n_starts: int, ) -> Result: """Fill a Result from a set of histories. Parameters ---------- problem: The problem to find optimal parameters for. n_starts: Number of performed multistarts. history_options: Optimizer history options. """ if history_options.storage_file is None: raise ValueError("No history file specified.") result = Result() result.problem = problem result.optimize_result = OptimizeResult() result.optimize_result.list = [ read_result_from_file(problem, history_options, str(istart)) for istart in range(n_starts) if os.path.exists(history_options.storage_file.format(id=str(istart))) ] if not result.optimize_result.list: logger.error("No history files found.") if len(result.optimize_result.list) != n_starts: logger.warning( f"History files were incomplete " f"({len(result.optimize_result.list)}/{n_starts})." ) result.optimize_result.sort() return result
[docs] def optimization_result_from_history( filename: str, problem: pypesto.Problem, ) -> Result: """Convert a saved hdf5 History to an optimization result. Used for interrupted optimization runs. Parameters ---------- filename: The name of the file in which the information are stored. problem: Problem, needed to identify what parameters to accept. Returns ------- A result object in which the optimization result is constructed from history. But missing "Time, Message and Exitflag" keys. """ result = Result() with h5py.File(filename, 'r') as f: ids = list(f[HISTORY].keys()) x0s = [f[f'{HISTORY}/{id}/{TRACE}/0/{X}'][()] for id in ids] for id, x0 in zip(ids, x0s): history = Hdf5History(id=id, file=filename) history.recover_options(filename) optimizer_history = OptimizerHistory( history=history, x0=x0, lb=problem.lb, ub=problem.ub, generate_from_history=True, ) optimizer_result = OptimizerResult(id=id) fill_result_from_history(optimizer_result, optimizer_history) result.optimize_result.append(optimizer_result) return result