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 separately 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.

Parameters:

Name	Type	Description	Default
input	dict	The input dictionary to evaluate.	required
**kwargs	dict	Additional keyword arguments to pass to the function.	{}

Returns:

Type	Description
dict	The evaluation result.

Source code in xopt/evaluator.py

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

    Further kwargs are passed to the function.

    Parameters
    ----------
    input : dict
        The input dictionary to evaluate.
    **kwargs : dict
        Additional keyword arguments to pass to the function.

    Returns
    -------
    dict
        The evaluation result.
    """
    return self.safe_function(input, **{**self.function_kwargs, **kwargs})

xopt.evaluator.Evaluator.evaluate_data

evaluate_data(input_data)

Evaluate a dataframe of inputs.

Parameters:

Name	Type	Description	Default
input_data	Union[DataFrame, List[Dict[str, float]], Dict[str, List[float]], Dict[str, float]]	The input data to evaluate.	required

Returns:

Type	Description
DataFrame	The evaluation results.

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 a dataframe of inputs.

    Parameters
    ----------
    input_data : Union[pd.DataFrame, List[Dict[str, float]], Dict[str, List[float]], Dict[str, float]]
        The input data to evaluate.

    Returns
    -------
    pd.DataFrame
        The evaluation results.
    """
    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_function_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.

Parameters:

Name	Type	Description	Default
*args	tuple	Positional arguments to pass to the function.	()
**kwargs	dict	Keyword arguments to pass to the function.	{}

Returns:

Type	Description
dict	The safe function outputs.

Source code in xopt/evaluator.py

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

    Parameters
    ----------
    *args : tuple
        Positional arguments to pass to the function.
    **kwargs : dict
        Keyword arguments to pass to the function.

    Returns
    -------
    dict
        The safe function outputs.
    """
    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	The input dictionary to submit.	required

Returns:

Type	Description
Future	The Future object representing the submitted task.

Source code in xopt/evaluator.py

def submit(self, input: Dict) -> Future:
    """
    Submit a single input to the executor.

    Parameters
    ----------
    input : dict
        The input dictionary to submit.

    Returns
    -------
    Future
        The Future object representing the submitted task.
    """
    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 class
    # 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 a dataframe of inputs to the executor.

Parameters:

Name	Type	Description	Default
input_data	DataFrame	The input data to submit.	required

Returns:

Type	Description
List[Future]	A list of Future objects representing the submitted tasks.

Source code in xopt/evaluator.py

def submit_data(self, input_data: pd.DataFrame) -> List[Future]:
    """
    Submit a dataframe of inputs to the executor.

    Parameters
    ----------
    input_data : pd.DataFrame
        The input data to submit.

    Returns
    -------
    List[Future]
        A list of Future objects representing the submitted tasks.
    """
    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

xopt.evaluator.Evaluator.validate_all

validate_all(values)

Validate all inputs before initializing the Evaluator.

Parameters:

Name	Type	Description	Default
values	dict	The input values to validate.	required

Returns:

Type	Description
dict	The validated input values.

Source code in xopt/evaluator.py

@model_validator(mode="before")
def validate_all(cls, values: Dict) -> Dict:
    """
    Validate all inputs before initializing the Evaluator.

    Parameters
    ----------
    values : dict
        The input values to validate.

    Returns
    -------
    dict
        The validated input values.
    """
    f = get_function(values["function"])
    kwargs = values.get("function_kwargs", {})
    kwargs = {**get_function_defaults(f), **kwargs}
    values["function"] = f
    values["function_kwargs"] = kwargs

    max_workers = values.pop("max_workers", 1)

    executor = values.pop("executor", None)
    if not executor:
        executor = (
            ProcessPoolExecutor(max_workers=max_workers)
            if max_workers > 1
            else DummyExecutor()
        )

    # Cast as a NormalExecutor
    values["executor"] = NormalExecutor[type(executor)](executor=executor)
    values["max_workers"] = max_workers

    return values