Evaluator

xopt.evaluator.Evaluator

Bases: XoptBaseModel

Xopt Evaluator for handling the parallel execution of an evaluate function.

Parameters:

Name	Type	Description	Default
function	Callable	Function to evaluate.	required
function_kwargs	dict	Any kwargs to pass on to this function.	{}
max_workers	int	Maximum number of workers.	1
executor	NormalExecutor	NormalExecutor or any instantiated Executor object	required
vectorized	bool	If true, lists of evaluation points will be sent to the evaluator function to be processed in parallel instead of evaluated seperately via mapping.	False

Functions

xopt.evaluator.Evaluator.evaluate

evaluate(input, **kwargs)

Evaluate a single input dict using Evaluator.function with Evaluator.function_kwargs.

Further kwargs are passed to the function.

Inputs: inputs: dict of inputs to be evaluated **kwargs: additional kwargs to pass to the function

Returns: function(input, **function_kwargs_updated)

Source code in xopt/evaluator.py

def evaluate(self, input: Dict, **kwargs):
    """
    Evaluate a single input dict using Evaluator.function with
    Evaluator.function_kwargs.

    Further kwargs are passed to the function.

    Inputs:
        inputs: dict of inputs to be evaluated
        **kwargs: additional kwargs to pass to the function

    Returns:
        function(input, **function_kwargs_updated)

    """
    return self.safe_function(input, **{**self.function_kwargs, **kwargs})

xopt.evaluator.Evaluator.evaluate_data

evaluate_data(input_data)

evaluate dataframe of inputs

Source code in xopt/evaluator.py

def evaluate_data(
    self,
    input_data: Union[
        pd.DataFrame,
        List[Dict[str, float]],
        Dict[str, List[float]],
        Dict[str, float],
    ],
) -> pd.DataFrame:
    """evaluate dataframe of inputs"""
    if self.vectorized:
        output_data = self.safe_function(input_data, **self.function_kwargs)
    else:
        # This construction is needed to avoid a pickle error
        # translate input data into pandas dataframes
        if not isinstance(input_data, DataFrame):
            try:
                input_data = DataFrame(input_data)
            except ValueError:
                input_data = DataFrame(input_data, index=[0])

        inputs = input_data.to_dict("records")

        funcs = [self.function] * len(inputs)
        kwargs = [self.function_kwargs] * len(inputs)

        output_data = self.executor.map(
            safe_function1_for_map,
            funcs,
            inputs,
            kwargs,
        )

    return pd.concat(
        [input_data, DataFrame(output_data, index=input_data.index)], axis=1
    )

xopt.evaluator.Evaluator.safe_function

safe_function(*args, **kwargs)

Safely call the function, handling exceptions.

Note that this should not be submitted to fuu

Source code in xopt/evaluator.py

def safe_function(self, *args, **kwargs):
    """
    Safely call the function, handling exceptions.

    Note that this should not be submitted to fuu
    """
    return safe_function(self.function, *args, **kwargs)

xopt.evaluator.Evaluator.submit

submit(input)

submit a single input to the executor

Parameters:

Name	Type	Description	Default
input	dict		required

Returns:

Name	Type	Description
Future	Future object

Source code in xopt/evaluator.py

def submit(self, input: Dict):
    """submit a single input to the executor

    Parameters
    ----------
    input : dict

    Returns
    -------
    Future  : Future object
    """
    if not isinstance(input, dict):
        raise ValueError("input must be a dictionary")
    # return self.executor.submit(self.function, input, **self.function_kwargs)
    # Must call a function outside of the classs
    # See: https://stackoverflow.com/questions/44144584/typeerror-cant-pickle-thread-lock-objects
    return self.executor.submit(
        safe_function, self.function, input, **self.function_kwargs
    )

xopt.evaluator.Evaluator.submit_data

submit_data(input_data)

submit dataframe of inputs to executor

Source code in xopt/evaluator.py

def submit_data(self, input_data: pd.DataFrame):
    """submit dataframe of inputs to executor"""
    input_data = pd.DataFrame(input_data)  # cast to dataframe for consistency

    if self.vectorized:
        # Single submission, cast to numpy array
        inputs = input_data.to_dict(orient="list")
        for key, value in inputs.items():
            inputs[key] = np.array(value)
        futures = [self.submit(inputs)]  # Single item
    else:
        # Do not use iterrows or itertuples.
        futures = [self.submit(inputs) for inputs in input_data.to_dict("records")]

    return futures