code: # -*- coding: utf-8 -*-
  """ core_model.py - define coure model API"""
  import os
  import abc
  import yaml
  import pandas as pd
  import numpy as np
  import logging
  import subprocess
  import warnings
  from contextlib import contextmanager
  from typing import Union, Mapping
  from ..workbench.em_framework.model import AbstractModel as AbstractWorkbenchModel
  from ..workbench.em_framework.evaluators import BaseEvaluator
  
  from typing import Collection
  from typing import Iterable
  
  from ..database.database import Database
  from ..scope.scope import Scope
  from ..optimization.optimization_result import OptimizationResult
  from ..optimization import EpsilonProgress, ConvergenceMetrics, SolutionCount
  from ..util.evaluators import prepare_evaluator
  from ..exceptions import MissingArchivePathError, ReadOnlyDatabaseError, MissingIdWarning
  
  from .._pkg_constants import *
  
  from ..util.loggers import get_module_logger
  _logger = get_module_logger(__name__)
  
  class AbstractCoreModel(abc.ABC, AbstractWorkbenchModel):
      """
      An interface for using a model with EMAT.
  
      Individual models should be instantiated using derived
      subclasses of this abstract base class, and not using
      this class directly.
  
      Args:
          configuration (str or Mapping or None):
              The configuration for this core model. This can be given
              explicitly as a `dict`, or as a `str` which gives the
              filename of a YAML file that will be loaded. If there is
              no configuration, giving `None` is also acceptable.
          scope (Scope or str):
              The exploration scope, as a `Scope` object or as
              a `str` which gives the filename of a YAML file that will be
              loaded.
          safe (bool):
              Load the configuration YAML file in 'safe' mode.
              This can be disabled if the configuration requires
              custom Python types or is otherwise not compatible with
              safe mode. Loading configuration files with safe mode
              off is not secure and should not be done with files from
              untrusted sources.
          db (Database, optional):
              An optional Database to store experiments and results.
          name (str, default "EMAT"):
              A name for this model, given as an alphanumeric string.
              The name is required by workbench operations.
          metamodel_id (int, optional):
              An identifier for this model, if it is a meta-model.
              Defaults to 0 (i.e., not a meta-model).
      """
  
      def __init__(self,
                   configuration:Union[str,Mapping,None],
                   scope,
                   safe=True,
                   db=None,
                   name='EMAT',
                   metamodel_id=0,
                   ):
          if isinstance(configuration, str):
              with open(configuration, 'r') as stream:
                  if safe:
                      configuration = yaml.safe_load(stream)
                  else:
                      configuration = yaml.load(stream, Loader=yaml.FullLoader)
              if configuration is None:
                  configuration = {}
  
          self.config = configuration if configuration is not None else {}
          self.db = db
          if isinstance(scope, Scope):
              self.scope = scope
          else:
              self.scope = Scope(scope)
  
          AbstractWorkbenchModel.__init__(self, name=name.replace('_','').replace(' ',''))
          self.uncertainties = self.scope._x_list
          self.levers = self.scope._l_list
          self.constants = self.scope._c_list
          self.outcomes = self.scope._m_list
  
          self.metamodel_id = metamodel_id
  
      def __getstate__(self):
          # don't pickle the db connection
          return dict((k, v) for (k, v) in self.__dict__.items() if (k != 'db'))
  
      @abc.abstractmethod
      def setup(self, params):
          """
          Configure the core model with the experiment variable values.
  
          This method is the place where the core model set up takes place,
          including creating or modifying files as necessary to prepare
          for a core model run.  When running experiments, this method
          is called once for each core model experiment, where each experiment
          is defined by a set of particular values for both the exogenous
          uncertainties and the policy levers.  These values are passed to
          the experiment only here, and not in the `run` method itself.
          This facilitates debugging, as the `setup` method can potentially
          be used without the `run` method, allowing the user to manually
          inspect the prepared files and ensure they are correct before
          actually running a potentially expensive model.
  
          Each input exogenous uncertainty or policy lever can potentially
          be used to manipulate multiple different aspects of the underlying
          core model.  For example, a policy lever that includes a number of
          discrete future network "build" options might trigger the replacement
          of multiple related network definition files.  Or, a single uncertainty
          relating to the cost of fuel might scale both a parameter linked to
          the modeled per-mile cost of operating an automobile, as well as the
          modeled total cost of fuel used by transit services.
  
          At the end of the `setup` method, a core model experiment should be
          ready to run using the `run` method.
  
          Args:
              params (dict):
                  experiment variables including both exogenous
                  uncertainty and policy levers
                  
          Raises:
              KeyError:
                  if a defined experiment variable is not supported
                  by the core model        
          """     
   
      @abc.abstractmethod
      def get_experiment_archive_path(
              self,
              experiment_id=None,
              makedirs=False,
              parameters=None,
              run_id=None,
      ):
          """
          Returns a file system location to store model run outputs.
  
          For core models with long model run times, it is recommended
          to store the complete model run results in an archive.  This
          will facilitate adding additional performance measures to the
          scope at a later time.
  
          Both the scope name and experiment id can be used to create the 
          folder path. 
          
          Args:
              experiment_id (int):
                  The experiment id, which is also the row id of the
                  experiment in the database. If this is omitted, an
                  experiment id is read or created using the parameters.
              makedirs (bool, default False):
                  If this archive directory does not yet exist, create it.
              parameters (dict, optional):
                  The parameters for this experiment, used to create or
                  lookup an experiment id. The parameters are ignored
                  if `experiment_id` is given.
              run_id (UUID, optional):
                  The run_id of this model run.  If not given but a
                  run_id attribute is stored in this FilesCoreModel
                  instance, that value is used.
  
          Returns:
              str: Experiment archive path (no trailing backslashes).
          """     
      
      @abc.abstractmethod
      def run(self):
          """
          Run the core model.
  
          This method is the place where the core model run takes place.
          Note that this method takes no arguments; all the input
          exogenous uncertainties and policy levers are delivered to the
          core model in the `setup` method, which will be executed prior
          to calling this method. This facilitates debugging, as the `setup`
          method can potentially be used without the `run` method, allowing
          the user to manually inspect the prepared files and ensure they
          are correct before actually running a potentially expensive model.
          When running experiments, this method is called once for each core
          model experiment, after the `setup` method completes.
  
          If the core model requires some post-processing by `post_process`
          method defined in this API, then when this function terminates
          the model directory should be in a state that is ready to run the
          `post_process` command next.
  
          Raises:
              UserWarning: If model is not properly setup
          """     
      
      def post_process(self, params, measure_names, output_path=None):
          """
          Runs post processors associated with particular performance measures.
  
          This method is the place to conduct automatic post-processing
          of core model run results, in particular any post-processing that
          is expensive or that will write new output files into the core model's
          output directory.  The core model run should already have
          been completed using `setup` and `run`.  If the relevant performance
          measures do not require any post-processing to create (i.e. they
          can all be read directly from output files created during the core
          model run itself) then this method does not need to be overloaded
          for a particular core model implementation.
  
          Args:
              params (dict):
                  Dictionary of experiment variables, with keys as variable names
                  and values as the experiment settings. Most post-processing
                  scripts will not need to know the particular values of the
                  inputs (exogenous uncertainties and policy levers), but this
                  method receives the experiment input parameters as an argument
                  in case one or more of these parameter values needs to be known
                  in order to complete the post-processing.
              measure_names (List[str]):
                  List of measures to be processed.  Normally for the first pass
                  of core model run experiments, post-processing will be completed
                  for all performance measures.  However, it is possible to use
                  this argument to give only a subset of performance measures to
                  post-process, which may be desirable if the post-processing
                  of some performance measures is expensive.  Additionally, this
                  method may also be called on archived model results, allowing
                  it to run to generate only a subset of (probably new) performance
                  measures based on these archived runs.
              output_path (str, optional):
                  Path to model outputs.  If this is not given (typical for the
                  initial run of core model experiments) then the local/default
                  model directory is used.  This argument is provided primarily
                  to facilitate post-processing archived model runs to make new
                  performance measures (i.e. measures that were not in-scope when
                  the core model was actually run).
  
          Raises:
              KeyError:
                  If post process is not available for specified measure
          """
      
      @abc.abstractmethod
      def load_measures(
              self,
              measure_names: Collection[str]=None,
              *,
              rel_output_path=None,
              abs_output_path=None,
      ) -> dict:
          """
          Import selected measures from the core model.
          
          This method is the place to put code that can actually reach into
          files in the core model's run results and extract performance
          measures. It is expected that it should not do any post-processing
          of results (i.e. it should read from but not write to the model
          outputs directory).
  
          Imports measures from active scenario
          
          Args:
              measure_names (Collection[str]):
                  Collection of measures to be loaded.
              rel_output_path, abs_output_path (str, optional):
                  Path to model output locations, either relative
                  to the `model_path` directory (when a subclass
                  is a type that has a model path) or as an absolute
                  directory.  If neither is given, the default
                  value is equivalent to setting `rel_output_path` to
                  'Outputs'.
  
          Returns:
              dict of measure name and values from active scenario
          
          Raises:
              KeyError: If load_measures is not available for specified
                  measure
          """           
          
  
      @abc.abstractmethod
      def archive(self, params, model_results_path, experiment_id:int=0):
          """
          Copies model outputs to archive location.
          
          Args:
              params (dict): Dictionary of experiment variables
              model_results_path (str): archive path
              experiment_id (int, optional): The id number for this experiment.
          
          """
  
      @property
      def allow_short_circuit(self):
          """
          Bool: Allow model runs to be skipped if measures already appear in the database.
          """
          return self.config.get('allow_short_circuit', True)
  
      @allow_short_circuit.setter
      def allow_short_circuit(self, value):
          self.config['allow_short_circuit'] = bool(value)
  
      @property
      def ignore_crash(self):
          """
          Bool: Allow model runs to `post_process` and `archive` even after an apparent crash in `run`.
          """
          return self.config.get('ignore_crash', False)
  
      @ignore_crash.setter
      def ignore_crash(self, value):
          self.config['ignore_crash'] = bool(value)
  
      @property
      def success_indicator(self):
          """
          str: The name of a file that indicates the model has run successfully.
  
          The flag is the mere existance of a file with this name, not any particular
          file content. This file is deleted automatically when the model `run` is
          initiated, so that it can be recreated to indicate a success.
          """
          return self.config.get('success_indicator', None)
  
      @success_indicator.setter
      def success_indicator(self, value):
          self.config['success_indicator'] = value
  
      @property
      def killed_indicator(self):
          """
          str: The name of a file that indicates the model was killed due to an unrecoverable error.
  
          The flag is the mere existance of a file with this name, not any particular
          file content. This file is deleted automatically when the model `run` is
          initiated, so that it can be recreated to indicate an unrecoverable error.
          """
          return self.config.get('killed_indicator', None)
  
      @killed_indicator.setter
      def killed_indicator(self, value):
          self.config['killed_indicator'] = value
  
      @property
      def local_directory(self):
          """Path: The current local working directory for this model."""
          return self.config.get("local_directory", os.getcwd())
  
      @local_directory.setter
      def local_directory(self, value):
          self.config["local_directory"] = value
  
      @property
      def resolved_model_path(self):
          """
          Path: The resolved model path.
  
          For core models that don't rely on the file system, this
          is set to the current working directory and is generally
          irrelevant. Overload this property for models that do
          rely on the file system.
          """
          return self.local_directory
  
      @property
      def is_db_locked(self):
          if self.db:
              return self.db.is_locked
          return False
  
      @contextmanager
      def lock_db(self, x=True):
          if x and self.db:
              with self.db.lock:
                  yield
          else:
              yield
  
      def enter_run_model(self):
          """A hook for actions at the very beginning of the run_model step."""
  
      def exit_run_model(self):
          """A hook for actions at the very end of the run_model step."""
  
      def run_model(self, scenario, policy):
          """
          Runs an experiment through core model.
  
          This method overloads the `run_model` method given in
          the EMA Workbench, and provides the correct execution
          of a core model within the workbench framework.  This
          function assembles and executes the steps laid out in
          other methods of this class, adding some useful logic
          to optimize the process (e.g. optionally short-
          circuiting runs that already have results stored
          in the database).
  
          For each experiment, the core model is called to:
  
              1.  `setup` experiment variables, copy files
                  as needed, and otherwise prepare to run the
                  core model for a particular experiment,
              2.  `run` the experiment,
              3.  `post_process` the result if needed to
                  produce all relevant performance measures,
              4.  `archive` model outputs from this experiment
                  (optional), and
              5.  `load_measures` from the experiment and
                  store those measures in the associated database.
  
          Note that this method does *not* return any outcomes.
          Outcomes are instead written into self.outcomes_output,
          and can be retrieved from there, or from the database at
          a later time.
  
          In general, it should not be necessary to overload this
          method in derived classes built for particular core models.
          Instead, write overloaded methods for `setup`, `run`,
          `post_process` , `archive`, and `load_measures`.  Moreover,
          in typical usage a modeler will generally not want to rely
          on this method directly, but instead use `run_experiments`
          to automatically run multiple experiments with one command.
  
          Args:
              scenario (Scenario): A dict-like object that
                  has key-value pairs for each uncertainty.
              policy (Policy): A dict-like object that
                  has key-value pairs for each lever.
  
          Raises:
              UserWarning: If there are no experiments associated with
                  this type.
  
          """
          self.enter_run_model()
          try:
              self.comment_on_run = None
  
              _logger.debug("run_core_model read_experiment_parameters")
  
              experiment_id = policy.get("_experiment_id_", None)
              if experiment_id is None:
                  experiment_id = scenario.get("_experiment_id_", None)
  
              if not hasattr(self, 'db') and hasattr(self, '_db'):
                  self.db = self._db
  
              # If running a core files model using the DistributedEvaluator,
              # the workers won't have access to the DB directly, so we'll only
              # run the short-circuit test and the ad-hoc write-to-database
              # section of this code if the `db` attribute is available.
              if hasattr(self, 'db') and self.db is not None:
  
                  assert isinstance(self.db, Database)
  
                  if experiment_id is None:
                      with warnings.catch_warnings():
                          if self.is_db_locked:
                              warnings.simplefilter("ignore", category=MissingIdWarning)
                          experiment_id = self.db.read_experiment_id(self.scope.name, scenario, policy)
  
                  if experiment_id and self.allow_short_circuit:
                      # opportunity to short-circuit run by loading pre-computed values.
                      precomputed = self.db.read_experiment_measures(
                          self.scope,
                          design_name=None,
                          experiment_id=experiment_id,
                      )
                      if not precomputed.empty:
                          self.outcomes_output = dict(precomputed.iloc[0])
                          self.log(f"short circuit experiment_id {experiment_id} / {getattr(self, 'uid', 'no uid')}")
                          return
  
                  if experiment_id is None and not self.is_db_locked:
                      experiment_id = self.db.write_experiment_parameters_1(
                          self.scope.name, 'ad hoc', scenario, policy
                      )
                  self.log(f"YES DATABASE experiment_id {experiment_id}", level=logging.DEBUG)
  
              else:
                  _logger.debug(f"NO DATABASE experiment_id {experiment_id}")
  
              xl = {}
              xl.update(scenario)
              xl.update(policy)
  
              m_names = self.scope.get_measure_names()
  
              _logger.debug(f"run_core_model setup {experiment_id}")
              self.setup(xl)
  
              if self.success_indicator is not None:
                  success_indicator = os.path.join(self.resolved_model_path, self.success_indicator)
                  if os.path.exists(success_indicator):
                      os.remove(success_indicator)
              else:
                  success_indicator = None
  
              if self.killed_indicator is not None:
                  killed_indicator = os.path.join(self.resolved_model_path, self.killed_indicator)
                  if os.path.exists(killed_indicator):
                      os.remove(killed_indicator)
              else:
                  killed_indicator = None
  
              _logger.debug(f"run_core_model run {experiment_id}")
              try:
                  self.run()
              except subprocess.CalledProcessError as err:
                  _logger.error(f"ERROR in run_core_model run {experiment_id}: {str(err)}")
                  try:
                      ex_archive_path = self.get_experiment_archive_path(experiment_id, makedirs=True)
                  except MissingArchivePathError:
                      pass
                  else:
                      if isinstance(err, subprocess.CalledProcessError):
                          if err.stdout:
                              with open(os.path.join(ex_archive_path, 'error.stdout.log'), 'ab') as stdout:
                                  stdout.write(err.stdout)
                          if err.stderr:
                              with open(os.path.join(ex_archive_path, 'error.stderr.log'), 'ab') as stderr:
                                  stderr.write(err.stderr)
                      with open(os.path.join(ex_archive_path, 'error.log'), 'a') as errlog:
                          errlog.write(str(err))
                  measures_dictionary = {name: np.nan for name in m_names}
                  # Assign to outcomes_output, for ema_workbench compatibility
                  self.outcomes_output = measures_dictionary
  
                  if not self.ignore_crash:
                      # If 'ignore_crash' is False (the default), then abort now and skip
                      # any post-processing and other archiving steps, which will
                      # probably fail anyway.
                      self.log(f"run_core_model ABORT {experiment_id}", level=logging.ERROR)
                      self.comment_on_run = f"FAILED EXPERIMENT {experiment_id}: {str(err)}"
                      return
                  else:
                      _logger.error(f"run_core_model CONTINUE AFTER ERROR {experiment_id}")
  
              try:
                  if success_indicator and not os.path.exists(success_indicator):
                      # The absence of the `success_indicator` file means that the model
                      # did not actually terminate correctly, so we do not want to
                      # post-process or store these results in the database.
                      self.comment_on_run = f"NON-SUCCESSFUL EXPERIMENT {experiment_id}: success_indicator missing"
                      raise ValueError(f"success_indicator missing: {success_indicator}")
  
                  if killed_indicator and os.path.exists(killed_indicator):
                      self.comment_on_run = f"KILLED EXPERIMENT {experiment_id}: killed_indicator present"
                      raise ValueError(f"killed_indicator present: {killed_indicator}")
  
                  _logger.debug(f"run_core_model post_process {experiment_id}")
                  self.post_process(xl, m_names)
  
                  _logger.debug(f"run_core_model wrap up {experiment_id}")
                  measures_dictionary = self.load_measures(m_names)
                  m_df = pd.DataFrame(measures_dictionary, index=[experiment_id])
  
              except KeyboardInterrupt:
                  _logger.exception(
                      f"KeyboardInterrupt in post_process, load_measures or outcome processing {experiment_id}")
                  raise
              except Exception as err:
                  _logger.exception(f"error in post_process, load_measures or outcome processing {experiment_id}")
                  _logger.error(f"proceeding directly to archive attempt {experiment_id}")
                  if not self.comment_on_run:
                      self.comment_on_run = f"PROBLEM IN EXPERIMENT {experiment_id}: {str(err)}"
              else:
                  # only write to database if there was no error in post_process, load_measures or outcome processing
                  if experiment_id and hasattr(self, 'db') and self.db is not None and not self.db.readonly:
                      _logger.debug(f"run_core_model write db {experiment_id}")
                      run_id = getattr(self, 'run_id', None)
                      if run_id is None:
                          run_id, _ = self.db.new_run_id(
                              scope_name=self.scope.name,
                              experiment_id=experiment_id,
                              source=self.metamodel_id or 0,
                          )
                      try:
                          self.db.write_experiment_measures(self.scope.name, self.metamodel_id, m_df, [run_id])
                      except ReadOnlyDatabaseError:
                          warnings.warn("database is read-only, not storing model outcomes")
                      except Exception as err:
                          _logger.exception(f"error in writing results to database: {str(err)}")
                      else:
                          _logger.debug(f"run_core_model OK write db {experiment_id} {self.metamodel_id} {run_id}\n{m_df}")
                  else:
                      _logger.debug(f"run_core_model no db to write to {experiment_id}")
  
              if experiment_id:
                  try:
                      ex_archive_path = self.get_experiment_archive_path(experiment_id)
                  except MissingArchivePathError:
                      pass
                  else:
                      _logger.debug(f"run_core_model archive {experiment_id}")
                      self.archive(xl, ex_archive_path, experiment_id)
              else:
                  _logger.debug(f"run_core_model no archive because no experiment_id")
          finally:
              self.exit_run_model()
  
      def read_experiments(
              self,
              design_name,
              db=None,
              only_pending=False,
              only_complete=False,
              only_with_measures=False,
      ):
          """
          Reads results from a design of experiments from the database.
  
          Args:
              design_name (str): The name of the design to load.
              db (Database, optional): The Database from which to read experiments.
                  If no db is given, the default `db` for this model is used.
              only_pending (bool, default False): If True, only pending
                  experiments (which have no performance measure results
                  stored in the database) are returned.
              only_complete (bool, default False): If True, only complete
                  experiments (which have no performance measure
                  results missing in the database) are returned.
              only_with_measures (bool, default False): If True, only
                  experiments with at least one stored performance measure
                  are returned.
  
          Returns:
              pandas.DataFrame:
                  A DataFrame that contains all uncertainties, levers, and measures
                  for the experiments.
  
          Raises:
              ValueError:
                  If there is no Database connection `db` set.
          """
          db = db if db is not None else self.db
          if db is None:
              raise ValueError('no database to read from')
  
          return self.ensure_dtypes(
              db.read_experiment_all(
                  self.scope.name,
                  design_name,
                  only_pending=only_pending,
                  only_complete=only_complete,
                  only_with_measures=only_with_measures,
              )
          )
  
      def read_experiment_parameters(
              self,
              design_name=None,
              db=None,
              only_pending=False,
              *,
              experiment_ids=None,
      ):
          """
          Reads uncertainties and levers from a design of experiments from the database.
  
          Args:
              design_name (str, optional): If given, only experiments
                  associated with both the scope and the named design
                  are returned, otherwise all experiments associated
                  with the scope are returned.
              db (Database, optional): The Database from which to read experiments.
                  If no db is given, the default `db` for this model is used.
              only_pending (bool, default False): If True, only pending
                  experiments (which have no performance measure results
                  stored in the database) are returned.
              experiment_ids (Collection, optional):
                  A collection of experiment id's to load.  If given,
                  both `design_name` and `only_pending` are ignored.
  
          Returns:
              pandas.DataFrame:
                  A DataFrame that contains all uncertainties, levers, and measures
                  for the experiments.
  
          Raises:
              ValueError:
                  If `db` is not given and there is no default
                  Database connection set.
          """
          db = db if db is not None else self.db
  
          if db is None:
              raise ValueError('no database to read from')
  
          return self.ensure_dtypes(
              db.read_experiment_parameters(
                  self.scope.name,
                  design_name,
                  only_pending=only_pending,
                  experiment_ids=experiment_ids,
              )
          )
  
      def read_experiment_measures(
              self,
              design_name,
              experiment_id=None,
              db=None,
      ):
          """
          Reads performance measures from a design of experiments from the database.
  
          Args:
              design_name (str): The name of the design to load.
              experiment_id (int, optional): The id of the experiment to load.
              db (Database, optional): The Database from which to read experiment(s).
                  If no db is given, the default `db` for this model is used.
  
          Returns:
              pandas.DataFrame:
                  A DataFrame that contains all uncertainties, levers, and measures
                  for the experiments.
  
          Raises:
              ValueError:
                  If `db` is not given and there is no default
                  Database connection set.
          """
          db = db if db is not None else self.db
  
          if db is None:
              raise ValueError('no database to read from')
  
          measures =  self.ensure_dtypes(
              db.read_experiment_measures(
                  self.scope.name,
                  design_name,
                  experiment_id,
                  source=self.metamodel_id,
              )
          )
          
          # only return measures within scope
          measures = measures[[i for i in self.scope.get_measure_names()
                               if i in measures.columns]]
          
          return measures
          
  
      def ensure_dtypes(self, df:pd.DataFrame):
          """
          Convert columns of dataframe to correct dtype as needed.
  
          Args:
              df (pandas.DataFrame): A dataframe with column names
                  that are uncertainties, levers, or measures.
  
          Returns:
              pandas.DataFrame:
                  The same data as input, but with dtypes as appropriate.
          """
          return self.scope.ensure_dtypes(df)
  
      def design_experiments(self, *args, **kwargs):
          """
          Create a design of experiments based on this model.
  
          Args:
              n_samples_per_factor (int, default 10): The number of samples in the
                  design per random factor.
              n_samples (int or tuple, optional): The total number of samples in the
                  design.  If `jointly` is False, this is the number of samples in each
                  of the uncertainties and the levers, the total number of samples will
                  be the square of this value.  Give a 2-tuple to set values for
                  uncertainties and levers respectively, to set them independently.
                  If this argument is given, it overrides `n_samples_per_factor`.
              random_seed (int or None, default 1234): A random seed for reproducibility.
              db (Database, optional): If provided, this design will be stored in the
                  database indicated.  If not provided, the `db` for this model will
                  be used, if one is set.
              design_name (str, optional): A name for this design, to identify it in the
                  database. If not given, a unique name will be generated based on the
                  selected sampler.
              sampler (str or AbstractSampler, default 'lhs'): The sampler to use for this
                  design.  Available pre-defined samplers include:
                      - 'lhs': Latin Hypercube sampling
                      - 'ulhs': Uniform Latin Hypercube sampling, which ignores defined
                          distribution shapes from the scope and samples everything
                          as if it was from a uniform distribution
                      - 'mc': Monte carlo sampling
                      - 'uni': Univariate sensitivity testing, whereby experiments are
                          generated setting each parameter individually to minimum and
                          maximum values (for numeric dtypes) or all possible values
                          (for boolean and categorical dtypes).  Note that designs for
                          univariate sensitivity testing are deterministic and the number
                          of samples given is ignored.
              sample_from ('all', 'uncertainties', or 'levers'): Which scope components
                  from which to sample.  Components not sampled are set at their default
                  values in the design.
              jointly (bool, default True): Whether to sample jointly all uncertainties
                  and levers in a single design, or, if False, to generate separate samples
                  for levers and uncertainties, and then combine the two in a full-factorial
                  manner.  This argument has no effect unless `sample_from` is 'all'.
                  Note that setting `jointly` to False may produce a very large design,
                  as the total number of experiments will be the product of the number of
                  experiments for the levers and the number of experiments for the
                  uncertainties, which are set separately (i.e. if `n_samples` is given,
                  the total number of experiments is the square of that value).
  
          Returns:
              pandas.DataFrame: The resulting design.
          """
          if 'scope' in kwargs:
              kwargs.pop('scope')
  
          if 'db' not in kwargs:
              kwargs['db'] = self.db
  
          from ..experiment import experimental_design
          return experimental_design.design_experiments(self.scope, *args, **kwargs)
  
      def async_experiments(
              self,
              design:pd.DataFrame=None,
              db=None,
              *,
              design_name=None,
              evaluator=None,
              max_n_workers=None,
              stagger_start=None,
              batch_size=None,
      ):
          """
          Asynchronously runs a design of combined experiments using this model.
  
          A combined experiment includes a complete set of input values for
          all exogenous uncertainties (a Scenario) and all policy levers
          (a Policy). Unlike the perform_experiments function in the EMA Workbench,
          this method pairs each Scenario and Policy in sequence, instead
          of running all possible combinations of Scenario and Policy.
          This change ensures compatibility with the EMAT database modules, which
          preserve the complete set of input information (both uncertainties
          and levers) for each experiment.  To conduct a full cross-factorial set
          of experiments similar to the default settings for EMA Workbench,
          use a factorial design, by setting the `jointly` argument for the
          `design_experiments` to False, or by designing experiments outside
          of EMAT with your own approach.
  
          Args:
              design (pandas.DataFrame, optional): experiment definitions
                  given as a DataFrame, where each exogenous uncertainties and
                  policy levers is given as a column, and each row is an experiment.
              db (Database, required): The database to use for loading and saving experiments.
                  If none is given, the default database for this model is used.
                  If there is no default db, and none is given here,
                  these experiments will be aborted.
              design_name (str, optional): The name of a design of experiments to
                  load from the database.  This design is only used if
                  `design` is None.
              evaluator (emat.workbench.Evaluator, optional): Optionally give an
                  evaluator instance.  If not given, a default DistributedEvaluator
                  will be instantiated.  Passing any other kind of evaluator will
                  currently cause an error, although in the future other async
                  compatible evaluators may be provided.
              max_n_workers (int, optional):
                  The maximum number of workers that will be created for a default
                  dask.distributed LocalCluster.  If the number of cores available is
                  smaller than this number, fewer workers will be spawned.  This value
                  is only used if a default LocalCluster has not yet been created.
              stagger_start (int, optional):
                  If provided, wait this number of seconds between initial dispatch
                  of experiments to the evaluator.  For models that do a lot of
                  file copying up front, this can prevent over-saturating the file
                  storage system.
              batch_size (int, optional):
                  For fast-running core models, the overhead from multi-processing
                  can represent a big chunk of overall runtime.  Grouping experiments
                  into batches that are sent to workers as a group can mitigate this.
                  Setting batch_size to 1 will process every experiment separately.
                  If no batch size is given, a guess is made as to an efficient
                  batch_size based on the number of experiments and the number of
                  workers.
  
          Raises:
              ValueError:
                  If there are no experiments defined.  This includes
                  the situation where `design` is given but no database is
                  available.
  
          """
          # catch user gives only a design, not experiment_parameters
          if isinstance(design, str) and design_name is None:
              design_name, design = design, None
  
          if design_name is None and design is None:
              raise ValueError(f"must give design_name or design")
  
          if db is None:
              db = self.db
  
          if design_name is not None and design is None:
              if not db:
                  raise ValueError(f'cannot load design "{design_name}", there is no db')
              design = db.read_experiment_parameters(self.scope.name, design_name)
  
          if design.empty:
              raise ValueError(f"no experiments available")
  
          from .asynchronous import  asynchronous_experiments
  
          if self.db is None:
              if db is not None:
                  self.db = db
              else:
                  raise ValueError("cannot run async_experiments without a `db` defined")
  
          return asynchronous_experiments(
              self,
              design,
              evaluator=evaluator,
              max_n_workers=max_n_workers,
              stagger_start=stagger_start,
              batch_size=batch_size,
          )
  
  
      def run_experiments(
              self,
              design=None,
              evaluator=None,
              *,
              design_name=None,
              db=None,
              allow_short_circuit=None,
      ):
          """
          Runs a design of combined experiments using this model.
  
          A combined experiment includes a complete set of input values for
          all exogenous uncertainties (a Scenario) and all policy levers
          (a Policy). Unlike the perform_experiments function in the EMA Workbench,
          this method pairs each Scenario and Policy in sequence, instead
          of running all possible combinations of Scenario and Policy.
          This change ensures compatibility with the EMAT database modules, which
          preserve the complete set of input information (both uncertainties
          and levers) for each experiment.  To conduct a full cross-factorial set
          of experiments similar to the default settings for EMA Workbench,
          use a factorial design, by setting the `jointly` argument for the
          `design_experiments` to False, or by designing experiments outside
          of EMAT with your own approach.
  
          Args:
              design (pandas.DataFrame, optional): experiment definitions
                  given as a DataFrame, where each exogenous uncertainty and
                  policy levers is given as a column, and each row is an experiment.
              evaluator (emat.workbench.Evaluator, optional): Optionally give an
                  evaluator instance.  If not given, a default SequentialEvaluator
                  will be instantiated.
              design_name (str, optional): The name of a design of experiments to
                  load from the database.  This design is only used if
                  `design` is None.
              db (Database, optional): The database to use for loading and saving experiments.
                  If none is given, the default database for this model is used.
                  If there is no default db, and none is given here,
                  the results are not stored in a database. Set to False to explicitly
                  not use the default database, even if it exists.
  
          Returns:
              pandas.DataFrame:
                  A DataFrame that contains all uncertainties, levers, and measures
                  for the experiments.
  
          Raises:
              ValueError:
                  If there are no experiments defined.  This includes
                  the situation where `design` is given but no database is
                  available.
  
          """
  
          from ..workbench import Scenario, Policy, perform_experiments
  
          # catch user gives only a design, not experiment_parameters
          if isinstance(design, str) and design_name is None:
              design_name, design = design, None
  
          if design_name is None and design is None:
              raise ValueError(f"must give design_name or design")
  
          if db is None:
              db = self.db
  
          if design_name is not None and design is None:
              if not db:
                  raise ValueError(f'cannot load design "{design_name}", there is no db')
              design = db.read_experiment_parameters(self.scope.name, design_name)
  
          if design.empty:
              raise ValueError(f"no experiments available")
  
          # catch metamodels here and run them as a batch, which is much faster
          function = getattr(self, 'function', None)
          from .meta_model import MetaModel
          if isinstance(function, MetaModel):
              outcomes = function.predict(design)
              result = self.ensure_dtypes(pd.concat([
                  design,
                  outcomes
              ], axis=1, sort=False))
              from ..experiment.experimental_design import ExperimentalDesign
              result = ExperimentalDesign(result)
              result.scope = self.scope
              result.design_name = getattr(design, 'design_name', None)
              result.sampler_name = getattr(design, 'sampler_name', None)
              if db:
                  metamodel_id = self.metamodel_id
                  if metamodel_id is None:
                      metamodel_id = db.get_new_metamodel_id(self.scope.name)
                  db.write_experiment_measures(self.scope.name, metamodel_id, outcomes)
              return result
  
          scenarios = []
          scenario_cols = self.scope._get_uncertainty_and_constant_names()
          design_scenarios = design[scenario_cols]
          for rownum in range(len(design)):
              if design.index.name == 'experiment':
                  s = Scenario(
                      _experiment_id_=design.index[rownum],
                      **design_scenarios.iloc[rownum],
                  )
              else:
                  s = Scenario(
                      _experiment_id_=False,
                      **design_scenarios.iloc[rownum],
                  )
              scenarios.append(s)
  
          lever_names = self.scope.get_lever_names()
          policies = [
              Policy(f"Incognito{n}", **dict(zip(lever_names, i)))
              for n,i in enumerate(design[lever_names].itertuples(index=False, name='ExperimentL'))
          ]
  
          evaluator = prepare_evaluator(evaluator, self)
  
          if getattr(evaluator, 'asynchronous', False):
              # When the evaluator is in asynchronous mode, the core model runs will be
              # dispatched here but the function will not block waiting on the result, and
              # instead depend on the model execution process to write the results into
              # the database when complete.
              with evaluator:
                  if allow_short_circuit is not None:
                      _stored_allow_short_circuit = self.allow_short_circuit
                      self.allow_short_circuit = allow_short_circuit
                  else:
                      _stored_allow_short_circuit = None
                  try:
                      perform_experiments(
                          self,
                          scenarios=scenarios,
                          policies=policies,
                          zip_over={'scenarios', 'policies'},
                          evaluator=evaluator,
                      )
                  finally:
                      if _stored_allow_short_circuit is not None:
                          self.allow_short_circuit = _stored_allow_short_circuit
              return
  
          else:
              with evaluator:
                  if db is False:
                      _stored_db = self.db
                      self.db = None
                  else:
                      _stored_db = None
                  if allow_short_circuit is not None:
                      _stored_allow_short_circuit = self.allow_short_circuit
                      self.allow_short_circuit = allow_short_circuit
                  else:
                      _stored_allow_short_circuit = None
                  try:
                      experiments, outcomes = perform_experiments(
                          self,
                          scenarios=scenarios,
                          policies=policies,
                          zip_over={'scenarios', 'policies'},
                          evaluator=evaluator,
                      )
                  finally:
                      if _stored_db:
                          self.db = _stored_db
                      if _stored_allow_short_circuit is not None:
                          self.allow_short_circuit = _stored_allow_short_circuit
              experiments.index = design.index
  
              outcomes = pd.DataFrame.from_dict(outcomes)
              outcomes.index = design.index
  
              # if db:
              #     metamodel_id = self.metamodel_id
              #     if metamodel_id is None:
              #         metamodel_id = 0
              #     db.write_experiment_measures(self.scope.name, metamodel_id, outcomes)
  
              # Put constants back into experiments
              experiments_ = experiments.drop(
                  columns=['scenario', 'policy', 'model', '_experiment_id_'],
                  errors='ignore',
              )
              for i in self.scope.get_constants():
                  experiments_[i.name] = i.value
  
              result = self.ensure_dtypes(pd.concat([
                  experiments_,
                  outcomes
              ], axis=1, sort=False))
              from ..experiment.experimental_design import ExperimentalDesign
              result = ExperimentalDesign(result)
              result.scope = self.scope
              result.design_name = getattr(design, 'design_name', None)
              result.sampler_name = getattr(design, 'sampler_name', None)
              return result
  
      def run_reference_experiment(
              self,
              evaluator=None,
              *,
              db=None,
      ):
          """
          Runs a reference experiment using this model.
  
          This single experiment includes a complete set of input values for
          all exogenous uncertainties (a Scenario) and all policy levers
          (a Policy). Each is set to the default value indicated by the scope.
  
          Args:
              evaluator (emat.workbench.Evaluator, optional): Optionally give an
                  evaluator instance.  If not given, a default SequentialEvaluator
                  will be instantiated.
              db (Database, optional): The database to use for loading and saving experiments.
                  If none is given, the default database for this model is used.
                  If there is no default db, and none is given here,
                  the results are not stored in a database. Set to False to explicitly
                  not use the default database, even if it exists.
  
          Returns:
              pandas.DataFrame:
                  A DataFrame that contains all uncertainties, levers, and measures
                  for the experiments.
  
          """
          if db is None:
              db = self.db
          ref = self.design_experiments(sampler='ref', db=db)
          return self.run_experiments(ref, evaluator=evaluator, db=db)
  
      def create_metamodel_from_data(
              self,
              experiment_inputs:pd.DataFrame,
              experiment_outputs:pd.DataFrame,
              output_transforms: dict = None,
              metamodel_id:int=None,
              include_measures=None,
              exclude_measures=None,
              db = None,
              random_state=None,
              experiment_stratification=None,
              suppress_converge_warnings=False,
              regressor = None,
              find_best_metamodeltype=False,
      ):
          """
          Create a MetaModel from a set of input and output observations.
  
          Args:
              experiment_inputs (pandas.DataFrame): This dataframe
                  should contain all of the experimental inputs, including
                  values for each uncertainty, level, and constant.
              experiment_outputs (pandas.DataFrame): This dataframe
                  should contain all of the experimental outputs, including
                  a column for each performance measure. The index
                  for the outputs should match the index for the
                  `experiment_inputs`, so that the I-O matches row-by-row.
              output_transforms (dict): Deprecated.  Specify the
                  output transforms directly in the scope instead.
              metamodel_id (int, optional): An identifier for this meta-model.
                  If not given, a unique id number will be created randomly.
              include_measures (Collection[str], optional): If provided, only
                  output performance measures with names in this set will be included.
              exclude_measures (Collection[str], optional): If provided, only
                  output performance measures with names not in this set will be included.
              db (Database, optional): The database to use for loading and saving metamodels.
                  If none is given, the default database for this model is used.
                  If there is no default db, and none is given here,
                  the metamodel is not stored in a database.
              random_state (int, optional): A random state to use in the metamodel
                  regression fitting.
              experiment_stratification (pandas.Series, optional):
                  A stratification of experiments, used in cross-validation.
              suppress_converge_warnings (bool, default False):
                  Suppress convergence warnings during metamodel fitting.
              regressor (Estimator, optional): A scikit-learn estimator implementing a
                  multi-target regression.  If not given, a detrended simple Gaussian
                  process regression is used.
              find_best_metamodeltype (int, default 0):
                  Run a search to find the best metamodeltype for each
                  performance measure, repeating each cross-validation
                  step this many times.  For more stable results, choose
                  3 or more, although larger numbers will be slow.  If
                  domain knowledge about the normal expected range and
                  behavior of each performance measure is available,
                  it is better to give the metamodeltype explicitly in
                  the Scope.
  
          Returns:
              MetaModel:
                  a callable object that, when called as if a
                  function, accepts keyword arguments as inputs and
                  returns a dictionary of (measure name: value) pairs.
          """
          from .meta_model import create_metamodel
  
          # The outputs index typically has a 2-level multi-index,
          # giving both experiment_id and run_id.  But for this
          # analysis, we will strip out the run_id.
          if experiment_outputs.index.nlevels == 2:
              experiment_outputs.index = experiment_outputs.index.get_level_values(0)
  
          return create_metamodel(
              scope=self.scope,
              experiments=pd.concat([experiment_inputs, experiment_outputs], axis=1),
              metamodel_id=metamodel_id,
              db=db,
              include_measures=include_measures,
              exclude_measures=exclude_measures,
              random_state=random_state,
              experiment_stratification=experiment_stratification,
              suppress_converge_warnings=suppress_converge_warnings,
              regressor=regressor,
              name=None,
              find_best_metamodeltype=find_best_metamodeltype,
          )
  
      def create_metamodel_from_design(
              self,
              design_name:str,
              metamodel_id:int = None,
              include_measures=None,
              exclude_measures=None,
              db=None,
              random_state=None,
              suppress_converge_warnings=False,
              regressor=None,
              find_best_metamodeltype=False,
      ):
          """
          Create a MetaModel from a set of input and output observations.
  
          Args:
              design_name (str): The name of the design to use.
              metamodel_id (int, optional): An identifier for this meta-model.
                  If not given, a unique id number will be created randomly.
              include_measures (Collection[str], optional): If provided, only
                  output performance measures with names in this set will be included.
              exclude_measures (Collection[str], optional): If provided, only
                  output performance measures with names not in this set will be included.
              random_state (int, optional): A random state to use in the metamodel
                  regression fitting.
              suppress_converge_warnings (bool, default False):
                  Suppress convergence warnings during metamodel fitting.
              regressor (Estimator, optional): A scikit-learn estimator implementing a
                  multi-target regression.  If not given, a detrended simple Gaussian
                  process regression is used.
              find_best_metamodeltype (int, default 0):
                  Run a search to find the best metamodeltype for each
                  performance measure, repeating each cross-validation
                  step this many times.  For more stable results, choose
                  3 or more, although larger numbers will be slow.  If
                  domain knowledge about the normal expected range and
                  behavior of each performance measure is available,
                  it is better to give the metamodeltype explicitly in
                  the Scope.
  
          Returns:
              MetaModel:
                  a callable object that, when called as if a
                  function, accepts keyword arguments as inputs and
                  returns a dictionary of (measure name: value) pairs.
  
          Raises:
              ValueError: If the named design still has pending experiments.
          """
          db = db if db is not None else self.db
  
          if db is None:
              raise ValueError("db is None")
  
          check_df = db.read_experiment_parameters(self.scope.name, design_name, only_pending=True)
          if not check_df.empty:
              from ..exceptions import PendingExperimentsError
              raise PendingExperimentsError(f'design "{design_name}" has pending experiments')
  
          experiment_inputs = db.read_experiment_parameters(self.scope.name, design_name)
          experiment_outputs = db.read_experiment_measures(self.scope.name, design_name)
  
          transforms = {
              i.name: i.metamodeltype
              for i in self.scope.get_measures()
          }
  
          return self.create_metamodel_from_data(
              experiment_inputs,
              experiment_outputs,
              transforms,
              metamodel_id=metamodel_id,
              include_measures=include_measures,
              exclude_measures=exclude_measures,
              db=db,
              random_state=random_state,
              suppress_converge_warnings=suppress_converge_warnings,
              regressor=regressor,
              find_best_metamodeltype=find_best_metamodeltype,
          )
  
      def create_metamodel_from_designs(
              self,
              design_names:str,
              metamodel_id:int = None,
              include_measures=None,
              exclude_measures=None,
              db=None,
              random_state=None,
              suppress_converge_warnings=False,
      ):
          """
          Create a MetaModel from multiple sets of input and output observations.
  
          Args:
              design_names (Collection[str]): The names of the designs to use.
              metamodel_id (int, optional): An identifier for this meta-model.
                  If not given, a unique id number will be created randomly.
              include_measures (Collection[str], optional): If provided, only
                  output performance measures with names in this set will be included.
              exclude_measures (Collection[str], optional): If provided, only
                  output performance measures with names not in this set will be included.
              random_state (int, optional): A random state to use in the metamodel
                  regression fitting.
              suppress_converge_warnings (bool, default False):
                  Suppress convergence warnings during metamodel fitting.
  
          Returns:
              MetaModel:
                  a callable object that, when called as if a
                  function, accepts keyword arguments as inputs and
                  returns a dictionary of (measure name: value) pairs.
  
          Raises:
              ValueError: If the named design still has pending experiments.
          """
          db = db if db is not None else self.db
  
          if db is not None:
              for design_name in design_names:
                  check_df = db.read_experiment_parameters(self.scope.name, design_name, only_pending=True)
                  if not check_df.empty:
                      from ..exceptions import PendingExperimentsError
                      raise PendingExperimentsError(f'design "{design_name}" has pending experiments')
  
          experiment_inputs = []
          for design_name in design_names:
              f = db.read_experiment_parameters(self.scope.name, design_name)
              f['_design_'] = design_name
              experiment_inputs.append(f)
          experiment_inputs = pd.concat(experiment_inputs)
  
          experiment_outputs = []
          for design_name in design_names:
              f =  db.read_experiment_measures(self.scope.name, design_name)
              # f['_design_'] = design_name
              experiment_outputs.append(f)
          experiment_outputs = pd.concat(experiment_outputs)
  
          transforms = {
              i.name: i.metamodeltype
              for i in self.scope.get_measures()
          }
  
          return self.create_metamodel_from_data(
              experiment_inputs.drop('_design_', axis=1),
              experiment_outputs,
              transforms,
              metamodel_id=metamodel_id,
              include_measures=include_measures,
              exclude_measures=exclude_measures,
              db=db,
              random_state=random_state,
              experiment_stratification=experiment_inputs['_design_'],
              suppress_converge_warnings=suppress_converge_warnings,
          )
  
  
      def feature_scores(
              self,
              design,
              return_type='styled',
              random_state=None,
              cmap='viridis',
              measures=None,
              shortnames=None,
      ):
          """
          Calculate feature scores based on a design of experiments.
  
          This method is provided as a convenient pass-through to the
          `feature_scores` function in the `analysis` sub-package, using
          the scope and database attached to this model.
  
          Args:
              design (str or pandas.DataFrame): The name of the design
                  of experiments to use for feature scoring, or a single
                  pandas.DataFrame containing the experimental design and
                  results.
              return_type ({'styled', 'figure', 'dataframe'}):
                  The format to return, either a heatmap figure as an SVG
                  render in and xmle.Elem, or a plain pandas.DataFrame,
                  or a styled dataframe.
              random_state (int or numpy.RandomState, optional):
                  Random state to use.
              cmap (string or colormap, default 'viridis'): matplotlib
                  colormap to use for rendering.
              measures (Collection, optional): The performance measures
                  on which feature scores are to be generated.  By default,
                  all measures are included.
  
          Returns:
              xmle.Elem or pandas.DataFrame:
                  Returns a rendered SVG as xml, or a DataFrame,
                  depending on the `return_type` argument.
  
          This function internally uses feature_scoring from the EMA Workbench, which in turn
          scores features using the "extra trees" regression approach.
          """
          from ..analysis.feature_scoring import feature_scores
          if shortnames is True:
              shortnames = self.scope
          return feature_scores(
              self.scope,
              design=design,
              return_type=return_type,
              db=self.db,
              random_state=random_state,
              cmap=cmap,
              measures=measures,
              shortnames=shortnames,
          )
  
      def get_feature_scores(self, *args, **kwargs):
          """
          Deprecated, use `Model.feature_scores`.
          """
          # for compatability with prior versions of TMIP-EMAT
          return self.feature_scores(*args, **kwargs)
  
      def _common_optimization_setup(
              self,
              epsilons=0.1,
              convergence='default',
              display_convergence=True,
              evaluator=None,
      ):
          import numbers
          if isinstance(epsilons, numbers.Number):
              epsilons = [epsilons]*len(self.outcomes)
  
          if convergence == 'default':
              convergence = ConvergenceMetrics(
                  EpsilonProgress(),
                  SolutionCount(),
              )
  
          if display_convergence and isinstance(convergence, ConvergenceMetrics):
              from IPython.display import display
              display(convergence)
  
          evaluator = prepare_evaluator(evaluator, self)
  
          return epsilons, convergence, display_convergence, evaluator
  
      def optimize(
              self,
              searchover='levers',
              evaluator=None,
              nfe=10000,
              convergence='default',
              display_convergence=True,
              convergence_freq=100,
              constraints=None,
              reference=None,
              reverse_targets=False,
              algorithm=None,
              epsilons='auto',
              min_epsilon=0.1,
              cache_dir=None,
              cache_file=None,
              check_extremes=False,
              **kwargs,
      ):
          """
          Perform multi-objective optimization over levers or uncertainties.
  
          The targets for the multi-objective optimization (i.e. whether each
          individual performance measures is to be maximized or minimized) are
          read from the model's scope.
  
          Args:
              searchover ({'levers', 'uncertainties'}):
                  Which group of inputs to search over.  The other group
                  will be set at their default values, unless other values
                  are provided in the `reference` argument.
              evaluator (Evaluator, optional): The evaluator to use to
                  run the model. If not given, a SequentialEvaluator will
                  be created.
              nfe (int, default 10_000): Number of function evaluations.
                  This generally needs to be fairly large to achieve stable
                  results in all but the most trivial applications.
              convergence ('default', None, or emat.optimization.ConvergenceMetrics):
                  A convergence display during optimization.  The default
                  value is to report the epsilon-progress (the number of
                  solutions that ever enter the candidate pool of non-dominated
                  solutions) and the number of solutions remaining in that candidate
                  pool.  Pass `None` explicitly to disable convergence tracking.
              display_convergence (bool, default True): Whether to automatically
                  display figures that dynamically track convergence.  Set to
                  `False` if you are not using this method within a Jupyter
                  interactive environment.
              convergence_freq (int, default 100): How frequently to update the
                  convergence measures.  There is some computational overhead to
                  these convergence updates, so setting a value too small may
                  noticeably slow down the process.
              constraints (Collection[Constraint], optional):
                  Solutions will be constrained to only include values that
                  satisfy these constraints. The constraints can be based on
                  the search parameters (levers or uncertainties, depending on the
                  value given for `searchover`), or performance measures, or
                  some combination thereof.
              reference (Mapping): A set of values for the non-active inputs,
                  i.e. the uncertainties if `searchover` is 'levers', or the
                  levers if `searchover` is 'uncertainties'.  Any values not
                  set here revert to the default values identified in the scope.
              reverse_targets (bool, default False): Whether to reverse the
                  optimization targets given in the scope (i.e., changing
                  minimize to maximize, or vice versa).  This will result in
                  the optimization searching for the worst outcomes, instead of
                  the best outcomes.
              algorithm (platypus.Algorithm, optional): Select an
                  algorithm for multi-objective optimization.  The default
                  algorithm is EpsNSGAII. See `platypus` documentation for details.
              epsilons (float or array-like): Used to limit the number of
                  distinct solutions generated.  Set to a larger value to get
                  fewer distinct solutions.
              cache_dir (path-like, optional): A directory in which to
                  cache results.  Most of the arguments will be hashed
                  to develop a unique filename for these results, making this
                  generally safer than `cache_file`.
              cache_file (path-like, optional): A file into which to
                  cache results.  If this file exists, the contents of the
                  file will be loaded and all other arguments are ignored.
                  Use with great caution.
              kwargs: Any additional arguments will be passed on to the
                  platypus algorithm.
  
          Returns:
              emat.OptimizationResult:
                  The set of non-dominated solutions found.
                  When `convergence` is given, the convergence measures are
                  included, as a pandas.DataFrame in the `convergence` attribute.
          """
          from ..util.disk_cache import load_cache_if_available, save_cache
          if isinstance(algorithm, str) or algorithm is None:
              alg = algorithm
          else:
              alg = algorithm.__name__
  
          if reference is not None:
              from ..workbench import Policy, Scenario
              if searchover == 'levers' and not isinstance(reference, Scenario):
                  reference = Scenario("ReferenceScenario", **reference)
              elif searchover == 'uncertainties' and not isinstance(reference, Policy):
                  reference = Policy("ReferencePolicy", **reference)
          else:
              if searchover == 'levers':
                  reference = self.scope.default_scenario()
              elif searchover == 'uncertainties':
                  reference = self.scope.default_policy()
  
          x, cache_file = load_cache_if_available(
              cache_file=cache_file,
              cache_dir=cache_dir,
              searchover=searchover,
              nfe=nfe,
              convergence=convergence,
              convergence_freq=convergence_freq,
              constraints=constraints,
              reference=reference,
              reverse_targets=reverse_targets,
              algorithm=alg,
              epsilons=epsilons,
          )
  
          if x is None:
              epsilons, convergence, display_convergence, evaluator = self._common_optimization_setup(
                  epsilons, convergence, display_convergence, evaluator
              )
  
              if reverse_targets:
                  for k in self.scope.get_measures():
                      k.kind_original = k.kind
                      k.kind = k.kind * -1
  
              _db_pause = self.db
  
              try:
                  self.db = None
                  with evaluator:
  
                      if epsilons == 'auto':
                          from ..workbench import perform_experiments
                          if searchover == 'levers':
                              _, trial_outcomes = perform_experiments(
                                  self,
                                  scenarios=reference,
                                  policies=30,
                                  evaluator=evaluator,
                              )
                          else:
                              _, trial_outcomes = perform_experiments(
                                  self,
                                  scenarios=30,
                                  policies=reference,
                                  evaluator=evaluator,
                              )
                          epsilons = [max(min_epsilon, np.std(trial_outcomes[mn]) / 20) for mn in self.scope.get_measure_names()]
  
                      results = evaluator.optimize(
                          searchover=searchover,
                          reference=reference,
                          nfe=nfe,
                          constraints=constraints,
                          convergence=convergence,
                          convergence_freq=convergence_freq,
                          epsilons=epsilons,
                          **kwargs,
                      )
  
                      if isinstance(results, tuple) and len(results) == 2:
                          results, result_convergence = results
                      else:
                          result_convergence = None
  
                      # Put constants back in to results
                      for i in self.scope.get_constants():
                          results[i.name] = i.value
  
                      results = self.ensure_dtypes(results)
                      x = OptimizationResult(results, result_convergence, scope=self.scope)
  
                      if searchover == 'levers':
                          x.scenarios = reference
                      elif searchover == 'uncertainties':
                          x.policies = reference
  
                      if check_extremes:
                          x.check_extremes(
                              self,
                              1 if check_extremes is True else check_extremes,
                              evaluator=evaluator,
                              searchover=searchover,
                              robust=False,
                          )
  
              finally:
                  if reverse_targets:
                      for k in self.scope.get_measures():
                          k.kind = k.kind_original
                          del k.kind_original
                  self.db = _db_pause
  
          elif display_convergence:
              _, convergence, display_convergence, _ = self._common_optimization_setup(
                  None, convergence, display_convergence, False
              )
              for c in convergence:
                  try:
                      c.rebuild(x.convergence)
                  except KeyboardInterrupt:
                      raise
                  except:
                      pass
  
          x.cache_file = cache_file
          save_cache(x, cache_file)
          return x
  
      def robust_optimize(
              self,
              robustness_functions,
              scenarios,
              evaluator=None,
              nfe=10000,
              convergence='default',
              display_convergence=True,
              convergence_freq=100,
              constraints=None,
              epsilons=0.1,
              cache_dir=None,
              cache_file=None,
              algorithm=None,
              check_extremes=False,
              **kwargs,
      ):
          """
          Perform robust optimization.
  
          The robust optimization generally a multi-objective optimization task.
          It is undertaken using statistical measures of outcomes evaluated across
          a number of scenarios, instead of using the individual outcomes themselves.
          For each candidate policy, the model is evaluated against all of the considered
          scenarios, and then the robustness measures are evaluated using the
          set of outcomes from the original runs.  The robustness measures
          are aggregate measures that are computed from a set of outcomes.
          For example, this may be expected value, median, n-th percentile,
          minimum, or maximum value of any individual outcome.  It is also
          possible to have joint measures, e.g. expected value of the larger
          of outcome 1 or outcome 2.
  
          Each robustness function is indicated as a maximization or minimization
          target, where higher or lower values are better, respectively.
          The optimization process then tries to identify one or more
          non-dominated solutions for the possible policy levers.
  
          Args:
              robustness_functions (Collection[Measure]): A collection of
                  aggregate statistical performance measures.
              scenarios (int or Collection): A collection of scenarios to
                  use in the evaluation(s), or give an integer to generate
                  that number of random scenarios.
              evaluator (Evaluator, optional): The evaluator to use to
                  run the model. If not given, a SequentialEvaluator will
                  be created.
              nfe (int, default 10_000): Number of function evaluations.
                  This generally needs to be fairly large to achieve stable
                  results in all but the most trivial applications.
              convergence ('default', None, or emat.optimization.ConvergenceMetrics):
                  A convergence display during optimization.
              display_convergence (bool, default True): Automatically display
                  the convergence metric figures when optimizing.
              convergence_freq (int, default 100): The frequency at which
                  convergence metric figures are updated.
              constraints (Collection[Constraint], optional)
                  Solutions will be constrained to only include values that
                  satisfy these constraints. The constraints can be based on
                  the policy levers, or on the computed values of the robustness
                  functions, or some combination thereof.
              epsilons (float or array-like): Used to limit the number of
                  distinct solutions generated.  Set to a larger value to get
                  fewer distinct solutions.
              cache_dir (path-like, optional): A directory in which to
                  cache results.  Most of the arguments will be hashed
                  to develop a unique filename for these results, making this
                  generally safer than `cache_file`.
              cache_file (path-like, optional): A file into which to
                  cache results.  If this file exists, the contents of the
                  file will be loaded and all other arguments are ignored.
                  Use with great caution.
              algorithm (platypus.Algorithm or str, optional): Select an
                  algorithm for multi-objective optimization.  The algorithm can
                  be given directly, or named in a string. See `platypus`
                  documentation for details.
              check_extremes (bool or int, default False): Conduct additional
                  evaluations, setting individual policy levers to their
                  extreme values, for each candidate Pareto optimal solution.
              kwargs: any additional arguments will be passed on to the
                  platypus algorithm.
  
          Returns:
              emat.OptimizationResult:
                  The set of non-dominated solutions found.
                  When `convergence` is given, the convergence measures are
                  included, as a pandas.DataFrame in the `convergence` attribute.
          """
          from ..optimization.optimize import robust_optimize
  
          from ..util.disk_cache import load_cache_if_available, save_cache
          if isinstance(algorithm, str) or algorithm is None:
              alg = algorithm
          else:
              alg = algorithm.__name__
          result, cache_file = load_cache_if_available(
              cache_file=cache_file,
              cache_dir=cache_dir,
              scenarios=scenarios,
              convergence=convergence,
              convergence_freq=convergence_freq,
              constraints=constraints,
              epsilons=epsilons,
              nfe=nfe,
              robustness_functions=robustness_functions,
              alg=alg,
              check_extremes=check_extremes,
          )
  
          if result is None:
              _db_pause = self.db
              try:
                  self.db = None
                  result = robust_optimize(
                      self,
                      robustness_functions,
                      scenarios,
                      evaluator=evaluator,
                      nfe=nfe,
                      convergence=convergence,
                      display_convergence=display_convergence,
                      convergence_freq=convergence_freq,
                      constraints=constraints,
                      epsilons=epsilons,
                      check_extremes=check_extremes,
                      **kwargs,
                  )
              finally:
                  self.db = _db_pause
          elif display_convergence:
              _, convergence, display_convergence, _ = self._common_optimization_setup(
                  None, convergence, display_convergence, False
              )
              for c in convergence:
                  try:
                      c.rebuild(result.convergence)
                  except KeyboardInterrupt:
                      raise
                  except:
                      pass
  
          result.cache_file = cache_file
          save_cache(result, cache_file)
          return result
  
      def robust_evaluate(
              self,
              robustness_functions,
              scenarios,
              policies,
              evaluator=None,
              cache_dir=None,
              suspend_db=True,
      ):
          """
          Perform robust evaluation(s).
  
          The robust evaluation is used to generate statistical measures
          of outcomes, instead of generating the individual outcomes themselves.
          For each policy, the model is evaluated against all of the considered
          scenarios, and then the robustness measures are evaluated using the
          set of outcomes from the original runs.  The robustness measures
          are aggregate measures that are computed from a set of outcomes.
          For example, this may be expected value, median, n-th percentile,
          minimum, or maximum value of any individual outcome.  It is also
          possible to have joint measures, e.g. expected value of the larger
          of outcome 1 or outcome 2.
  
          Args:
              robustness_functions (Collection[Measure]): A collection of
                  aggregate statistical performance measures.
              scenarios (int or Collection): A collection of scenarios to
                  use in the evaluation(s), or give an integer to generate
                  that number of random scenarios.
              policies (int, or collection): A collection of policies to
                  use in the evaluation(s), or give an integer to generate
                  that number of random policies.
              evaluator (Evaluator, optional): The evaluator to use to
                  run the model. If not given, a SequentialEvaluator will
                  be created.
              cache_dir (path-like, optional): A directory in which to
                  cache results.
              suspend_db (bool, default True):
                  Suspend writing the results of individual model runs to
                  the database.  Robust evaluation potentially generates a
                  large number of model executions, and storing all these
                  individual results may not be useful.
  
          Returns:
              pandas.DataFrame: The computed value of each item
                  in `robustness_functions`, for each policy in `policies`.
          """
          robust_results = None
          cache_file = None
          if cache_dir is not None:
              try:
                  from ..util.hasher import hash_it
                  hh = hash_it(
                      scenarios,
                      policies,
                      robustness_functions,
                  )
                  os.makedirs(os.path.join(cache_dir,hh[2:4],hh[4:6]), exist_ok=True)
                  cache_file = os.path.join(cache_dir,hh[2:4],hh[4:6],hh[6:]+".gz")
                  if os.path.exists(cache_file):
                      _logger.debug(f"loading from cache_file={cache_file}")
                      from ..util.filez import load
                      robust_results = load(cache_file)
                      cache_file = None
              except KeyboardInterrupt:
                  raise
              except:
                  import traceback
                  warnings.warn('unable to manage cache')
                  traceback.print_exc()
  
          if robust_results is None:
              with self.lock_db(suspend_db):
                  if evaluator is None:
                      from ..workbench.em_framework import SequentialEvaluator
                      evaluator = SequentialEvaluator(self)
  
                  if not isinstance(evaluator, BaseEvaluator):
                      from dask.distributed import Client
                      if isinstance(evaluator, Client):
                          from ..workbench.em_framework.ema_distributed import DistributedEvaluator
                          evaluator = DistributedEvaluator(self, client=evaluator)
  
                  from ..workbench.em_framework.samplers import sample_uncertainties, sample_levers
  
                  if isinstance(scenarios, int):
                      n_scenarios = scenarios
                      scenarios = sample_uncertainties(self, n_scenarios)
  
                  with evaluator:
                      robust_results = evaluator.robust_evaluate(
                          robustness_functions,
                          scenarios,
                          policies,
                      )
  
                  robust_results = self.ensure_dtypes(robust_results)
  
          if cache_file is not None:
              from ..util.filez import save
              save(robust_results, cache_file, overwrite=True)
              with open(cache_file.replace('.gz','.info.txt'), 'wt') as notes:
                  print("scenarios=", scenarios, file=notes)
                  print("robustness_functions=", robustness_functions, file=notes)
                  print("policies=", policies, file=notes)
  
          return robust_results
  
      def io_experiment(self, params):
          """
          Run an experiment, and return a dictionary of inputs and outputs together.
  
          Args:
              params: dict
  
          Returns:
              dict
          """
          out = self.run_experiment(params).copy()
          out.update(params)
          return out
  
      def log(self, message, level=logging.INFO):
          """
          Log a message.
  
          This facility will attempt to send log messages to
          the attached database, falling back to the regular
          module logger in case that fails.
  
          Args:
              message (str): Message to send to log.
              level (int, default logging.INFO): Log level.
  
          Returns:
  
          """
          db = getattr(self, 'db', None)
          try:
              db.log(message, level=level)
          except:
              _logger.log(level, message)
  
apis: numpy.std pandas.DataFrame pandas.concat pandas.DataFrame.from_dict