Xopt

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
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
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
stopping_condition	StoppingCondition	An optional stopping condition to check during optimization. If provided, the optimization will stop when this condition is met.	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.

Source code in xopt/base.py

class Xopt(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
    ----------
    generator : SerializeAsAny[Generator]
        An object responsible for generating candidates for optimization.
    evaluator : SerializeAsAny[Evaluator]
        An object used for evaluating candidates generated by the generator.
    strict : bool, optional
        A flag indicating whether exceptions raised during evaluation should stop the
        optimization process.
    dump_file : str, optional
        An optional file path for dumping attributes of the xopt object and the
        results of evaluations.
    data : DataFrame, optional
        An optional DataFrame object for storing internal data related to the optimization
        process.
    serialize_torch : bool
        A flag indicating whether Torch (PyTorch) models should be serialized when
        saving them.
    serialize_inline : bool
        A flag indicating whether Torch models should be stored via binary string
        directly inside the main configuration file.
    stopping_condition : StoppingCondition, optional
        An optional stopping condition to check during optimization. If provided,
        the optimization will stop when this condition is met.

    Methods
    -------
    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(input_dict: dict)
        Evaluates a candidate without storing data.
    evaluate_data(input_data)
        Evaluates a set of candidates, adding the results to the internal DataFrame.
    add_data(new_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(n_samples=1, seed=None, **kwargs)
        Generates random inputs using the VOCS and evaluates them, adding the data to
        Xopt.
    yaml(**kwargs)
        Serializes the Xopt configuration to a YAML string.
    dump(file: str = None, **kwargs)
        Dumps the Xopt configuration to a specified file.
    dict(**kwargs) -> dict
        Provides a custom dictionary representation of the Xopt configuration.
    json(**kwargs) -> str
        Serializes the Xopt configuration to a JSON string.
    """

    generator: Union[SerializeAsAny[Generator], Any] = Field(
        description="generator object for Xopt"
    )
    evaluator: SerializeAsAny[Evaluator] = Field(
        description="evaluator object for Xopt"
    )
    strict: bool = Field(
        True,
        description="flag to indicate if exceptions raised during evaluation "
        "should stop Xopt",
    )
    dump_file: Optional[str] = Field(
        None, description="file to dump the results of the evaluations"
    )
    data: Optional[DataFrame] = Field(None, description="internal DataFrame object")
    serialize_torch: bool = Field(
        False,
        description="flag to indicate that torch models should be serialized "
        "when dumping",
    )
    serialize_inline: bool = Field(
        False,
        description="flag to indicate if torch models"
        " should be stored inside main config file",
    )
    stopping_condition: Optional[StoppingConditionUnion] = Field(
        None,
        description="optional stopping condition to check during optimization",
    )

    @model_validator(mode="before")
    @classmethod
    def validate_generator_and_legacy_vocs(cls, data: Any):
        """
        Validate the Xopt model by checking the generator and evaluator.
        """
        if isinstance(data, dict):
            # Handle Xopt 2.x style VOCS definition
            if "vocs" in data.keys():
                generator = data["generator"]

                # Move 2.x VOCS definition into generator dict definition if able
                if isinstance(generator, dict):
                    if "vocs" in generator:
                        raise VOCSError(
                            "Duplicate VOCS definitions. Please only define under `generator`."
                        )
                    else:
                        warnings.warn(
                            "Defining VOCS in the base Xopt object is deprecated and support will be removed from Xopt in an upcoming release. Please define it in the generator object instead.",
                            DeprecationWarning,
                            stacklevel=2,
                        )
                        generator["vocs"] = data.pop("vocs")

                # Drop VOCS definition in base if user is creating object via 2.x python API. Note: this replicates the behavior
                # from Xopt 2.x where the VOCS object in `generator` is the one which takes precedence.
                elif isinstance(generator, Generator):
                    warnings.warn(
                        "Defining VOCS in the base Xopt object is deprecated and support will be removed from Xopt in an upcoming release. Please remove the definition from `Xopt` and pass to generator object only.",
                        DeprecationWarning,
                        stacklevel=2,
                    )
                    data.pop("vocs")

                # Probably shouldn't end up here, but error out if `vocs` exists and `generator` isn't
                # a dict or a `Generator`.
                else:
                    raise VOCSError(
                        "Defining VOCS in the base Xopt object is deprecated. Please define in generator object only."
                    )

            # validate generator
            if isinstance(data["generator"], dict):
                name = data["generator"].pop("name")
                generator_class = get_generator(name)
                data["generator"] = generator_class.model_validate(data["generator"])

            # make a copy of the generator / vocs objects to avoid modifying the original
            data["generator"] = deepcopy(data["generator"])

        return data

    @field_validator("evaluator", mode="before")
    def validate_evaluator(cls, value):
        if isinstance(value, dict):
            value = Evaluator(**value)

        return value

    @field_validator("data", mode="before")
    def validate_data(cls, v, info: ValidationInfo):
        if isinstance(v, dict):
            try:
                v = pd.DataFrame(v)
                v.index = v.index.astype(np.int64)
                v = v.sort_index()
            except IndexError:
                v = pd.DataFrame(v, index=[0])
        elif isinstance(v, DataFrame):
            if not pd.api.types.is_integer_dtype(v.index):
                raise ValueError("dataframe index must be integer")
        # also add data to generator
        # TODO: find a more robust way of doing this
        generator = info.data["generator"]

        # Some generators need to maintain their own state (such as sequential generators)
        if not isinstance(generator, StateOwner):
            generator.add_data(v)
        else:
            generator.set_data(v)

        return v

    @model_validator(mode="before")
    @classmethod
    def max_evaluations_legacy(cls, data: Any):
        """
        Handle backward compatibility: convert old max_evaluations parameter
        to the new MaxEvaluationsCondition stopping condition.
        """
        if isinstance(data, dict) and "max_evaluations" in data:
            warnings.warn(
                "The attribute `max_evaluations` in `Xopt` is deprecated and will be removed in later versions of the Xopt "
                "library. Please use `stopping_condition` instead.",
                DeprecationWarning,
                stacklevel=2,
            )
            if data.get("stopping_condition") is not None:
                raise ValueError(
                    "Cannot specify both 'max_evaluations' and 'stopping_condition'. "
                    "Use 'stopping_condition' with MaxEvaluationsCondition instead."
                )
            max_evals = data.pop("max_evaluations")
            data["stopping_condition"] = MaxEvaluationsCondition(
                max_evaluations=max_evals
            )
        return data

    @property
    def n_data(self) -> int:
        if self.data is None:
            return 0
        else:
            return len(self.data)

    @property
    def vocs(self) -> VOCS:
        """
        Get the VOCS object from the generator.

        Returns
        -------
        VOCS
            The VOCS object associated with the generator.
        """
        if self.generator is None:
            raise ValueError("generator is not set")

        return self.generator.vocs

    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)

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

        Notes
        -----
        This method performs the following steps:
        (1) Determines the number of candidates to request from the generator.
        (2) Passes the candidate request to the generator.
        (3) Submits candidates to the evaluator.
        (4) Waits until all evaluations are finished
        (5) 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.suggest(n_generate)

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

    def run(self):
        """
        Run until the stopping criteria are met.

        Stops when any of the following conditions are met:
        1. Stopping condition is met (if stopping_condition is set)
        2. Generator is done
        """
        logger.info("Running Xopt")

        # Require stopping criterion
        if self.stopping_condition is None:
            raise ValueError("stopping_condition must be set to call Xopt.run()")

        while True:
            # Check custom stopping condition
            if self.stopping_condition is not None:
                if self.data is not None and self.stopping_condition.should_stop(
                    self.data, self.vocs
                ):
                    logger.info("Xopt is done. Stopping condition met.")
                    break

            self.step()

        # at the end, call the finalize method for the generator
        self.generator.finalize()

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

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

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

        """
        inputs = deepcopy(input_dict)

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

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

    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")
        validate_input_data(self.vocs, input_data)

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

        # if we are using a sequential generator that is active, make sure that the evaluated data matches the last candidate
        if isinstance(self.generator, SequentialGenerator):
            if self.generator.is_active:
                self.generator.validate_point(input_data)

        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

    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
        # Pass data to generator, continue in case of invalid data when strict=False
        try:
            self.generator.ingest(new_data.to_dict(orient="records"))
        except DataError as exc:
            if self.strict:
                raise exc

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

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

    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

    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.

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

    def grid_evaluate(
        self,
        n_samples: Union[int, dict[str, int]],
        custom_bounds: dict = None,
    ):
        """
        Evaluate a meshgrid of points using the VOCS and add the results to the internal
        DataFrame.

        Parameters
        ----------
        n_samples : int or dict
            The number of samples along each axis to evaluate on a meshgrid.
            If an int is provided, the same number of samples is used for all axes.
        custom_bounds : dict, optional
            dictionary of vocs-like ranges for mesh sampling.

        Returns
        -------
        pd.DataFrame
            The results of the evaluations added to the internal DataFrame.
        """
        gi = grid_inputs(
            self.vocs, n_samples, custom_bounds=custom_bounds, include_constants=True
        )
        result = self.evaluate_data(gi)
        return result

    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)

    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}")

    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)
        if not isinstance(result["generator"], dict):  # may return as module.path
            result["generator"] = {"name": result["generator"]}
        result["generator"] = {"name": get_generator_name(self.generator)} | result[
            "generator"
        ]
        return result

    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)
        if not isinstance(dict_result["generator"], dict):  # may return as module.path
            dict_result["generator"] = {"name": dict_result["generator"]}
        dict_result["generator"] = {
            "name": get_generator_name(self.generator)
        } | dict_result["generator"]
        dict_result["data"] = (
            json.loads(self.data.to_json()) if self.data is not None else None
        )

        if "stopping_condition" in dict_result:
            if dict_result["stopping_condition"] is not None:
                dict_result["stopping_condition"] = {
                    "name": self.stopping_condition.__class__.__name__
                } | dict_result["stopping_condition"]

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

        return json.dumps(dict_result)

    def __repr__(self):
        """
        Return information about the Xopt object, including the YAML representation
        without data.

        Returns
        -------
        str
            A string representation of the Xopt object.

        """
        # lazy import to avoid circular import
        from xopt import __version__

        # get dict minus data
        config = json.loads(self.json())
        config.pop("data")
        return f"""
            Xopt
________________________________
Version: {__version__}
Data size: {self.n_data}
Config as YAML:
{yaml.dump(config)}
"""

    def __str__(self):
        """
        Return a string representation of the Xopt object.

        Returns
        -------
        str
            A string representation of the Xopt object.

        """
        return self.__repr__()

vocs property

Get the VOCS object from the generator.

Returns:

Type	Description
VOCS	The VOCS object associated with the generator.

__init__(*args, **kwargs)

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)

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
    # Pass data to generator, continue in case of invalid data when strict=False
    try:
        self.generator.ingest(new_data.to_dict(orient="records"))
    except DataError as exc:
        if self.strict:
            raise exc

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)
    if not isinstance(result["generator"], dict):  # may return as module.path
        result["generator"] = {"name": result["generator"]}
    result["generator"] = {"name": get_generator_name(self.generator)} | result[
        "generator"
    ]
    return result

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}")

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
dict	The result of the evaluation.

Source code in xopt/base.py

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

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

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

    """
    inputs = deepcopy(input_dict)

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

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

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[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")
    validate_input_data(self.vocs, input_data)

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

    # if we are using a sequential generator that is active, make sure that the evaluated data matches the last candidate
    if isinstance(self.generator, SequentialGenerator):
        if self.generator.is_active:
            self.generator.validate_point(input_data)

    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

grid_evaluate(n_samples, custom_bounds=None)

Evaluate a meshgrid of points using the VOCS and add the results to the internal DataFrame.

Parameters:

Name	Type	Description	Default
n_samples	int or dict	The number of samples along each axis to evaluate on a meshgrid. If an int is provided, the same number of samples is used for all axes.	required
custom_bounds	dict	dictionary of vocs-like ranges for mesh sampling.	None

Returns:

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

Source code in xopt/base.py

def grid_evaluate(
    self,
    n_samples: Union[int, dict[str, int]],
    custom_bounds: dict = None,
):
    """
    Evaluate a meshgrid of points using the VOCS and add the results to the internal
    DataFrame.

    Parameters
    ----------
    n_samples : int or dict
        The number of samples along each axis to evaluate on a meshgrid.
        If an int is provided, the same number of samples is used for all axes.
    custom_bounds : dict, optional
        dictionary of vocs-like ranges for mesh sampling.

    Returns
    -------
    pd.DataFrame
        The results of the evaluations added to the internal DataFrame.
    """
    gi = grid_inputs(
        self.vocs, n_samples, custom_bounds=custom_bounds, include_constants=True
    )
    result = self.evaluate_data(gi)
    return result

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)
    if not isinstance(dict_result["generator"], dict):  # may return as module.path
        dict_result["generator"] = {"name": dict_result["generator"]}
    dict_result["generator"] = {
        "name": get_generator_name(self.generator)
    } | dict_result["generator"]
    dict_result["data"] = (
        json.loads(self.data.to_json()) if self.data is not None else None
    )

    if "stopping_condition" in dict_result:
        if dict_result["stopping_condition"] is not None:
            dict_result["stopping_condition"] = {
                "name": self.stopping_condition.__class__.__name__
            } | dict_result["stopping_condition"]

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

    return json.dumps(dict_result)

max_evaluations_legacy(data) classmethod

Handle backward compatibility: convert old max_evaluations parameter to the new MaxEvaluationsCondition stopping condition.

Source code in xopt/base.py

@model_validator(mode="before")
@classmethod
def max_evaluations_legacy(cls, data: Any):
    """
    Handle backward compatibility: convert old max_evaluations parameter
    to the new MaxEvaluationsCondition stopping condition.
    """
    if isinstance(data, dict) and "max_evaluations" in data:
        warnings.warn(
            "The attribute `max_evaluations` in `Xopt` is deprecated and will be removed in later versions of the Xopt "
            "library. Please use `stopping_condition` instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        if data.get("stopping_condition") is not None:
            raise ValueError(
                "Cannot specify both 'max_evaluations' and 'stopping_condition'. "
                "Use 'stopping_condition' with MaxEvaluationsCondition instead."
            )
        max_evals = data.pop("max_evaluations")
        data["stopping_condition"] = MaxEvaluationsCondition(
            max_evaluations=max_evals
        )
    return data

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.

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

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

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()

run()

Run until the stopping criteria are met.

Stops when any of the following conditions are met: 1. Stopping condition is met (if stopping_condition is set) 2. Generator is done

Source code in xopt/base.py

def run(self):
    """
    Run until the stopping criteria are met.

    Stops when any of the following conditions are met:
    1. Stopping condition is met (if stopping_condition is set)
    2. Generator is done
    """
    logger.info("Running Xopt")

    # Require stopping criterion
    if self.stopping_condition is None:
        raise ValueError("stopping_condition must be set to call Xopt.run()")

    while True:
        # Check custom stopping condition
        if self.stopping_condition is not None:
            if self.data is not None and self.stopping_condition.should_stop(
                self.data, self.vocs
            ):
                logger.info("Xopt is done. Stopping condition met.")
                break

        self.step()

    # at the end, call the finalize method for the generator
    self.generator.finalize()

step()

Run one optimization cycle.

Notes

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

Source code in xopt/base.py

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

    Notes
    -----
    This method performs the following steps:
    (1) Determines the number of candidates to request from the generator.
    (2) Passes the candidate request to the generator.
    (3) Submits candidates to the evaluator.
    (4) Waits until all evaluations are finished
    (5) 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.suggest(n_generate)

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

validate_generator_and_legacy_vocs(data) classmethod

Validate the Xopt model by checking the generator and evaluator.

Source code in xopt/base.py

@model_validator(mode="before")
@classmethod
def validate_generator_and_legacy_vocs(cls, data: Any):
    """
    Validate the Xopt model by checking the generator and evaluator.
    """
    if isinstance(data, dict):
        # Handle Xopt 2.x style VOCS definition
        if "vocs" in data.keys():
            generator = data["generator"]

            # Move 2.x VOCS definition into generator dict definition if able
            if isinstance(generator, dict):
                if "vocs" in generator:
                    raise VOCSError(
                        "Duplicate VOCS definitions. Please only define under `generator`."
                    )
                else:
                    warnings.warn(
                        "Defining VOCS in the base Xopt object is deprecated and support will be removed from Xopt in an upcoming release. Please define it in the generator object instead.",
                        DeprecationWarning,
                        stacklevel=2,
                    )
                    generator["vocs"] = data.pop("vocs")

            # Drop VOCS definition in base if user is creating object via 2.x python API. Note: this replicates the behavior
            # from Xopt 2.x where the VOCS object in `generator` is the one which takes precedence.
            elif isinstance(generator, Generator):
                warnings.warn(
                    "Defining VOCS in the base Xopt object is deprecated and support will be removed from Xopt in an upcoming release. Please remove the definition from `Xopt` and pass to generator object only.",
                    DeprecationWarning,
                    stacklevel=2,
                )
                data.pop("vocs")

            # Probably shouldn't end up here, but error out if `vocs` exists and `generator` isn't
            # a dict or a `Generator`.
            else:
                raise VOCSError(
                    "Defining VOCS in the base Xopt object is deprecated. Please define in generator object only."
                )

        # validate generator
        if isinstance(data["generator"], dict):
            name = data["generator"].pop("name")
            generator_class = get_generator(name)
            data["generator"] = generator_class.model_validate(data["generator"])

        # make a copy of the generator / vocs objects to avoid modifying the original
        data["generator"] = deepcopy(data["generator"])

    return data

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)