Xopt

xopt.Xopt

Xopt(*args, **kwargs)

Bases: XoptBaseModel

Object to handle a single optimization problem.

Xopt is designed for managing a single optimization problem by unifying the definition, configuration, and execution of optimization tasks. It combines the Variables, Objective, Constraints, Statics (VOCS) definition with a generator for candidate generation and an evaluator for objective function evaluations.

Parameters:

Name	Type	Description	Default
vocs	VOCS	VOCS object for defining the problem's variables, objectives, constraints, and statics.	required
generator	SerializeAsAny[Generator]	An object responsible for generating candidates for optimization.	required
evaluator	SerializeAsAny[Evaluator]	An object used for evaluating candidates generated by the generator.	required
strict	bool	A flag indicating whether exceptions raised during evaluation should stop the optimization process.	required
dump_file	str	An optional file path for dumping attributes of the xopt object and the results of evaluations.	required
max_evaluations	int	An optional maximum number of evaluations to perform. If set, the optimization process will stop after reaching this limit.	required
data	DataFrame	An optional DataFrame object for storing internal data related to the optimization process.	required
serialize_torch	bool	A flag indicating whether Torch (PyTorch) models should be serialized when saving them.	required
serialize_inline	bool	A flag indicating whether Torch models should be stored via binary string directly inside the main configuration file.	required

Methods:

Name	Description
step	Executes one optimization cycle, generating candidates, submitting them for evaluation, waiting for evaluation results, and updating data storage.
run	Runs the optimization process until the specified stopping criteria are met, such as reaching the maximum number of evaluations.
evaluate	Evaluates a candidate without storing data.
evaluate_data	Evaluates a set of candidates, adding the results to the internal DataFrame.
add_data	Adds new data to the internal DataFrame and the generator's data.
reset_data	Resets the internal data by clearing the DataFrame.
random_evaluate	Generates random inputs using the VOCS and evaluates them, adding the data to Xopt.
yaml	Serializes the Xopt configuration to a YAML string.
dump	Dumps the Xopt configuration to a specified file.
dict	Provides a custom dictionary representation of the Xopt configuration.
json	Serializes the Xopt configuration to a JSON string.

Initialize Xopt.

Parameters:

Name	Type	Description	Default
args	tuple	Positional arguments; a single YAML string can be passed as the only argument to initialize Xopt.	()
kwargs	dict	Keyword arguments for initializing Xopt.	{}

Raises:

Type	Description
ValueError	If both a YAML string and keyword arguments are specified during initialization. If more than one positional argument is provided.

Notes

• If a single YAML string is provided in the args argument, it is deserialized into keyword arguments using yaml.safe_load.
• When using the YAML string for initialization, no additional keyword arguments are allowed.

Source code in xopt/base.py

def __init__(self, *args, **kwargs):
    """
    Initialize Xopt.

    Parameters
    ----------
    args : tuple
        Positional arguments; a single YAML string can be passed as the only argument
        to initialize Xopt.
    kwargs : dict
        Keyword arguments for initializing Xopt.

    Raises
    ------
    ValueError
        If both a YAML string and keyword arguments are specified during
        initialization.
        If more than one positional argument is provided.

    Notes
    -----
    - If a single YAML string is provided in the `args` argument, it is deserialized
      into keyword arguments using `yaml.safe_load`.
    - When using the YAML string for initialization, no additional keyword arguments
      are allowed.

    """
    if len(args) == 1:
        if len(kwargs) > 0:
            raise ValueError("cannot specify yaml string and kwargs for Xopt init")
        super().__init__(**yaml.safe_load(args[0]))
    elif len(args) > 1:
        raise ValueError(
            "arguments to Xopt must be either a single yaml string "
            "or a keyword arguments passed directly to pydantic"
        )
    else:
        super().__init__(**kwargs)

Functions

xopt.Xopt.add_data

add_data(new_data)

Concatenate new data to the internal DataFrame and add it to the generator's data.

Parameters:

Name	Type	Description	Default
new_data	DataFrame	New data to be added to the internal DataFrame.	required

Source code in xopt/base.py

def add_data(self, new_data: pd.DataFrame):
    """
    Concatenate new data to the internal DataFrame and add it to the generator's
    data.

    Parameters
    ----------
    new_data : pd.DataFrame
        New data to be added to the internal DataFrame.

    """
    logger.debug(f"Adding {len(new_data)} new data to internal dataframes")

    # Set internal dataframe.
    if self.data is not None:
        new_data = pd.DataFrame(new_data, copy=True)  # copy for reindexing
        new_data.index = np.arange(len(self.data), len(self.data) + len(new_data))

        self.data = pd.concat([self.data, new_data], axis=0)
    else:
        if new_data.index.dtype != np.int64:
            new_data.index = new_data.index.astype(np.int64)
        self.data = new_data
    self.generator.add_data(new_data)

xopt.Xopt.dict

dict(**kwargs)

Handle custom dictionary generation.

Parameters:

Name	Type	Description	Default
**kwargs		Additional keyword arguments for customizing the dictionary generation.	{}

Returns:

Type	Description
Dict	A dictionary representation of the Xopt configuration.

Source code in xopt/base.py

def dict(self, **kwargs) -> Dict:
    """
    Handle custom dictionary generation.

    Parameters
    ----------
    **kwargs
        Additional keyword arguments for customizing the dictionary generation.

    Returns
    -------
    Dict
        A dictionary representation of the Xopt configuration.

    """
    result = super().model_dump(**kwargs)
    result["generator"] = {"name": self.generator.name} | result["generator"]
    return result

xopt.Xopt.dump

dump(file=None, **kwargs)

Dump data to a file.

Parameters:

Name	Type	Description	Default
file	str	The path to the file where the Xopt configuration will be dumped.	None
**kwargs		Additional keyword arguments for customizing the dump.	{}

Raises:

Type	Description
ValueError	If no dump file is specified via argument or in the dump_file attribute.

Source code in xopt/base.py

def dump(self, file: str = None, **kwargs):
    """
    Dump data to a file.

    Parameters
    ----------
    file : str, optional
        The path to the file where the Xopt configuration will be dumped.
    **kwargs
        Additional keyword arguments for customizing the dump.

    Raises
    ------
    ValueError
        If no dump file is specified via argument or in the `dump_file` attribute.

    """
    fname = file if file is not None else self.dump_file

    if fname is None:
        raise ValueError(
            "no dump file specified via argument or in `dump_file` attribute"
        )
    else:
        with open(fname, "w") as f:
            f.write(self.yaml(**kwargs))
        logger.debug(f"Dumped state to YAML file: {fname}")

xopt.Xopt.evaluate

evaluate(input_dict)

Evaluate a candidate without storing data.

Parameters:

Name	Type	Description	Default
input_dict	Dict	A dictionary representing the input data for candidate evaluation.	required

Returns:

Type	Description
Any	The result of the evaluation.

Source code in xopt/base.py

def evaluate(self, input_dict: Dict):
    """
    Evaluate a candidate without storing data.

    Parameters
    ----------
    input_dict : Dict
        A dictionary representing the input data for candidate evaluation.

    Returns
    -------
    Any
        The result of the evaluation.

    """
    inputs = deepcopy(input_dict)

    # add constants to input data
    for name, value in self.vocs.constants.items():
        inputs[name] = value

    self.vocs.validate_input_data(DataFrame(inputs, index=[0]))
    return self.evaluator.evaluate(input_dict)

xopt.Xopt.evaluate_data

evaluate_data(input_data)

Evaluate data using the evaluator and wait for results.

This method evaluates a set of candidates and adds the results to the internal DataFrame.

Parameters:

Name	Type	Description	Default
input_data	Union[pd.DataFrame, List[Dict[str, float], Dict[str, List[float],	Dict[str, float]]] The input data for evaluation, which can be provided as a DataFrame, a list of dictionaries, or a single dictionary.	required

Returns:

Type	Description
DataFrame	The results of the evaluations added to the internal DataFrame.

Source code in xopt/base.py

def evaluate_data(
    self,
    input_data: Union[
        pd.DataFrame,
        List[Dict[str, float]],
        Dict[str, List[float]],
        Dict[str, float],
    ],
) -> pd.DataFrame:
    """
    Evaluate data using the evaluator and wait for results.

    This method evaluates a set of candidates and adds the results to the internal
    DataFrame.

    Parameters
    ----------
    input_data : Union[pd.DataFrame, List[Dict[str, float], Dict[str, List[float],
                    Dict[str, float]]]
        The input data for evaluation, which can be provided as a DataFrame, a list of
        dictionaries, or a single dictionary.

    Returns
    -------
    pd.DataFrame
        The results of the evaluations added to the internal DataFrame.

    """
    # translate input data into pandas dataframes
    if not isinstance(input_data, DataFrame):
        try:
            input_data = DataFrame(deepcopy(input_data))
        except ValueError:
            input_data = DataFrame(deepcopy(input_data), index=[0])

    logger.debug(f"Evaluating {len(input_data)} inputs")
    self.vocs.validate_input_data(input_data)

    # add constants to input data
    for name, value in self.vocs.constants.items():
        input_data[name] = value

    output_data = self.evaluator.evaluate_data(input_data)

    if self.strict:
        validate_outputs(output_data)

    # explode any list like results if all the output names exist
    output_data = explode_all_columns(output_data)

    self.add_data(output_data)

    # dump data to file if specified
    if self.dump_file is not None:
        self.dump()

    return output_data

xopt.Xopt.json

json(**kwargs)

Handle custom serialization of generators and DataFrames.

Parameters:

Name	Type	Description	Default
**kwargs		Additional keyword arguments for customizing serialization.	{}

Returns:

Type	Description
str	The Xopt configuration serialized as a JSON string.

Source code in xopt/base.py

def json(self, **kwargs) -> str:
    """
    Handle custom serialization of generators and DataFrames.

    Parameters
    ----------
    **kwargs
        Additional keyword arguments for customizing serialization.

    Returns
    -------
    str
        The Xopt configuration serialized as a JSON string.

    """
    result = super().to_json(**kwargs)
    dict_result = json.loads(result)
    dict_result["generator"] = {"name": self.generator.name} | dict_result[
        "generator"
    ]
    dict_result["data"] = (
        json.loads(self.data.to_json()) if self.data is not None else None
    )

    # TODO: implement version checking
    # dict_result["xopt_version"] = __version__

    return json.dumps(dict_result)

xopt.Xopt.random_evaluate

random_evaluate(n_samples=None, seed=None, custom_bounds=None)

Convenience method to generate random inputs using VOCs and evaluate them.

This method generates random inputs using the Variables, Objectives, Constraints, and Statics (VOCS) and evaluates them, adding the data to the Xopt object and generator.

Parameters:

Name	Type	Description	Default
n_samples	int	The number of random samples to generate.	None
seed	int	The random seed for reproducibility.	None
custom_bounds	dict	Dictionary of vocs-like ranges for random sampling	None

Returns:

Type	Description
DataFrame	The results of the evaluations added to the internal DataFrame.

Source code in xopt/base.py

def random_evaluate(
    self,
    n_samples=None,
    seed=None,
    custom_bounds: dict = None,
):
    """
    Convenience method to generate random inputs using VOCs and evaluate them.

    This method generates random inputs using the Variables, Objectives,
    Constraints, and Statics (VOCS) and evaluates them, adding the data to the
    Xopt object and generator.

    Parameters
    ----------
    n_samples : int, optional
        The number of random samples to generate.
    seed : int, optional
        The random seed for reproducibility.
    custom_bounds : dict, optional
        Dictionary of vocs-like ranges for random sampling


    Returns
    -------
    pd.DataFrame
        The results of the evaluations added to the internal DataFrame.

    """
    random_inputs = self.vocs.random_inputs(
        n_samples, seed=seed, custom_bounds=custom_bounds, include_constants=True
    )
    result = self.evaluate_data(random_inputs)
    return result

xopt.Xopt.remove_data

remove_data(indices, inplace=True)

Removes data from the X.data data storage attribute.

Parameters:

Name	Type	Description	Default
indices	list[int]	List of indices specifying the rows (steps) to remove from data.	required
inplace	bool	Whether to update data inplace. If False, returns a copy.	True

Returns:

Type	Description
DataFrame or None	A copy of the internal DataFrame with the specified rows removed or None if inplace is True.

Source code in xopt/base.py

def remove_data(
    self, indices: list[int], inplace: bool = True
) -> Optional[pd.DataFrame]:
    """
    Removes data from the `X.data` data storage attribute.

    Parameters
    ----------
    indices: list of integers
        List of indices specifying the rows (steps) to remove from data.

    inplace: boolean, optional
        Whether to update data inplace. If False, returns a copy.

    Returns
    -------
    pd.DataFrame or None
        A copy of the internal DataFrame with the specified rows removed
        or None if inplace is True.

    """
    new_data = self.data.drop(labels=indices)
    new_data.index = np.arange(len(new_data), dtype=np.int64)
    if inplace:
        self.data = new_data
        self.generator.data = new_data
    else:
        return new_data

xopt.Xopt.reset_data

reset_data()

Reset the internal data by clearing the DataFrame.

Source code in xopt/base.py

def reset_data(self):
    """
    Reset the internal data by clearing the DataFrame.

    """
    self.data = pd.DataFrame()
    self.generator.data = pd.DataFrame()

xopt.Xopt.run

run()

Run until the maximum number of evaluations is reached or the generator is done.

Source code in xopt/base.py

def run(self):
    """
    Run until the maximum number of evaluations is reached or the generator is done.

    """
    while not self.generator.is_done:
        # Stopping criteria
        if self.max_evaluations is not None:
            if self.n_data >= self.max_evaluations:
                logger.info(
                    "Xopt is done. "
                    f"Max evaluations {self.max_evaluations} reached."
                )
                break

        self.step()

xopt.Xopt.step

step()

Run one optimization cycle.

This method performs the following steps: - Determines the number of candidates to request from the generator. - Passes the candidate request to the generator. - Submits candidates to the evaluator. - Waits until all evaluations are finished - Updates data storage and generator data storage (if applicable).

Source code in xopt/base.py

def step(self):
    """
    Run one optimization cycle.

    This method performs the following steps:
    - Determines the number of candidates to request from the generator.
    - Passes the candidate request to the generator.
    - Submits candidates to the evaluator.
    - Waits until all evaluations are finished
    - Updates data storage and generator data storage (if applicable).

    """
    logger.info("Running Xopt step")

    # get number of candidates to generate
    n_generate = self.evaluator.max_workers

    # generate samples and submit to evaluator
    logger.debug(f"Generating {n_generate} candidates")
    new_samples = self.generator.generate(n_generate)

    if new_samples is not None:
        # Evaluate data
        self.evaluate_data(new_samples)

xopt.Xopt.yaml

yaml(**kwargs)

Serialize the Xopt configuration to a YAML string.

Parameters:

Name	Type	Description	Default
**kwargs		Additional keyword arguments for customizing serialization.	{}

Returns:

Type	Description
str	The Xopt configuration serialized as a YAML string.

Source code in xopt/base.py

def yaml(self, **kwargs):
    """
    Serialize the Xopt configuration to a YAML string.

    Parameters
    ----------
    **kwargs
        Additional keyword arguments for customizing serialization.

    Returns
    -------
    str
        The Xopt configuration serialized as a YAML string.

    """
    output = json.loads(
        self.json(
            serialize_torch=self.serialize_torch,
            serialize_inline=self.serialize_inline,
            **kwargs,
        )
    )
    return yaml.dump(output)