Skip to content

Sequential Generator Base Class

SequentialGenerator

Bases: Generator, StateOwner

A generator that runs a sequential optimization algorithm, such as Nelder-Mead, extremum seeking, RCDS, etc.

Generally, these algorithms need an internal state to keep track of the optimization process. Additionally, users cannot interrupt the optimization process to add new points. These algorithms will start from the last point in the history.

Source code in xopt/generators/sequential/sequential_generator.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
class SequentialGenerator(Generator, StateOwner):
    """
    A generator that runs a sequential optimization algorithm, such as Nelder-Mead, extremum seeking, RCDS, etc.

    Generally, these algorithms need an internal state to keep track of the optimization process.
    Additionally, users cannot interrupt the optimization process to add new points.
    These algorithms will start from the last point in the history.
    """

    is_active: bool = False
    _last_candidate: Optional[List[Dict[str, float]]] = None
    _data_set: bool = False

    def add_data(self, new_data: pd.DataFrame):
        """
        Add new data to the generator.

        Parameters
        ----------
        new_data : pd.DataFrame
            The new data to add.

        Raises
        ------
        ValueError
            If the generator is active but no candidate was generated, or if the new data does not contain the last candidate.
        """
        # if the generator is active then the new data must contain the last candidate
        if self.is_active:
            if self._last_candidate is None:
                raise SeqGeneratorError(
                    "Generator is active, but no candidate was generated. Cannot add data."
                )
            if len(new_data) > 1:
                raise SeqGeneratorError(
                    "Cannot add more than one data point when generator is active."
                )
            else:
                # check if the last candidate is in the new data
                self.validate_point(new_data.iloc[0].to_dict())

        # do not call super, this will likely need to be customized for some generators
        if self.data is not None:
            self.data = pd.concat([self.data, new_data], axis=0, ignore_index=True)
        else:
            self.data = new_data

        # update internal state of the generator
        self._add_data(new_data)

    def validate_point(self, point: Dict[str, float]):
        """determine if an input point was generated by the generator"""
        last_candidate = np.array(
            [self._last_candidate[0][ele] for ele in self.vocs.variable_names]
        )
        point_variables = np.array(
            [point[ele] for ele in self.vocs.variable_names]
        ).flatten()
        if not np.allclose(last_candidate, point_variables, atol=0.0, rtol=1e-6):
            raise SeqGeneratorError(
                "Cannot add data that was not generated by the generator when generator is active. "
                "Call reset() to reset the generator first in order to add data via other methods."
            )

    @abstractmethod
    def _add_data(self, new_data: pd.DataFrame):
        """
        Customization of adding data to the algorithm.

        Parameters
        ----------
        new_data : pd.DataFrame
            The new data to add.

        Raises
        ------
        NotImplementedError
            If the method is not implemented by the subclass.
        """
        pass  # pragma: no cover

    def set_data(self, data: pd.DataFrame):
        """
        Set the full dataset for the generator. Typically only used when loading from a save file. This skips active
        generator lockout.

        Parameters
        ----------
        data : pd.DataFrame
            The data to set.
        """
        # TODO: make a flag for generator that support multiple data sets
        if self._data_set:
            raise SeqGeneratorError(
                "Data has already been initialized for this generator."
            )
        self._set_data(data)
        self._data_set = True

    @abstractmethod
    def _set_data(self, data: pd.DataFrame):
        pass  # pragma: no cover

    def generate(self, n_candidates: int = 1) -> dict:
        """
        Generate a new candidate point.

        Parameters
        ----------
        n_candidates : int, optional
            Number of candidates to generate, by default 1.

        Returns
        -------
        dict
            A dictionary representing the candidate point.

        Raises
        ------
        ValueError
            If more than one candidate is requested.
        """
        # we cannot generate more than one candidate at a time
        if n_candidates > 1:
            raise SeqGeneratorError(
                "Sequential generators can only generate one candidate at a time."
            )

        # if the generator is not active, we need to start it
        if not self.is_active:
            candidate = self._generate(True)
            self.is_active = True
        else:
            candidate = self._generate()

        # need to store the candidate to validate adding data to the generator
        self._last_candidate = candidate

        return candidate

    @abstractmethod
    def _generate(self, first_gen: bool = False) -> Optional[List[Dict[str, float]]]:
        """
        Generate a new candidate point.

        Returns
        -------
        dict
            A dictionary representing the candidate point.

        Raises
        ------
        NotImplementedError
            If the method is not implemented by the subclass.
        """
        pass  # pragma: no cover

    def reset(self):
        """
        Reset the generator.
        """
        self.is_active = False
        self._last_candidate = None
        self._reset()

    def _reset(self):
        """
        Reset the generator.

        """
        pass  # pragma: no cover

    def _get_initial_point(self) -> Tuple[np.ndarray, np.ndarray]:
        """
        Get the initial x0, f0 value from data.

        Returns
        -------
        np.ndarray
            The initial point as a NumPy array.
        np.ndarray
            The corresponding function value as a NumPy array.

        Raises
        ------
        ValueError
            If there are no points in the data.
        """
        # get the initial x0 value from data
        if self.data is None or len(self.data) == 0:
            raise ValueError(
                f"At least one point is required to start {self.__class__.__name__}, add data manually or via Xopt.random_evaluate() or Xopt.evaluate_data()"
            )
        x0 = self.data.iloc[-1][self.vocs.variable_names].to_numpy(dtype=float)
        f0 = self.data.iloc[-1][self.vocs.objective_names].to_numpy(dtype=float)
        return x0, f0

__init__(**kwargs)

Initialize the generator.

Source code in xopt/generator.py 
119
120
121
122
123
124
def __init__(self, **kwargs):
    """
    Initialize the generator.
    """
    super().__init__(**kwargs)
    logger.info(f"Initialized generator {self.name}")

add_data(new_data)

Add new data to the generator.

Parameters:

Name Type Description Default
new_data DataFrame

The new data to add.

 required

Raises:

Type Description
ValueError

If the generator is active but no candidate was generated, or if the new data does not contain the last candidate.
Source code in xopt/generators/sequential/sequential_generator.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def add_data(self, new_data: pd.DataFrame):
    """
    Add new data to the generator.

    Parameters
    ----------
    new_data : pd.DataFrame
        The new data to add.

    Raises
    ------
    ValueError
        If the generator is active but no candidate was generated, or if the new data does not contain the last candidate.
    """
    # if the generator is active then the new data must contain the last candidate
    if self.is_active:
        if self._last_candidate is None:
            raise SeqGeneratorError(
                "Generator is active, but no candidate was generated. Cannot add data."
            )
        if len(new_data) > 1:
            raise SeqGeneratorError(
                "Cannot add more than one data point when generator is active."
            )
        else:
            # check if the last candidate is in the new data
            self.validate_point(new_data.iloc[0].to_dict())

    # do not call super, this will likely need to be customized for some generators
    if self.data is not None:
        self.data = pd.concat([self.data, new_data], axis=0, ignore_index=True)
    else:
        self.data = new_data

    # update internal state of the generator
    self._add_data(new_data)

generate(n_candidates=1)

Generate a new candidate point.

Parameters:

Name Type Description Default
n_candidates int

Number of candidates to generate, by default 1.

 1

Returns:

Type Description
dict

A dictionary representing the candidate point.

Raises:

Type Description
ValueError

If more than one candidate is requested.
Source code in xopt/generators/sequential/sequential_generator.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def generate(self, n_candidates: int = 1) -> dict:
    """
    Generate a new candidate point.

    Parameters
    ----------
    n_candidates : int, optional
        Number of candidates to generate, by default 1.

    Returns
    -------
    dict
        A dictionary representing the candidate point.

    Raises
    ------
    ValueError
        If more than one candidate is requested.
    """
    # we cannot generate more than one candidate at a time
    if n_candidates > 1:
        raise SeqGeneratorError(
            "Sequential generators can only generate one candidate at a time."
        )

    # if the generator is not active, we need to start it
    if not self.is_active:
        candidate = self._generate(True)
        self.is_active = True
    else:
        candidate = self._generate()

    # need to store the candidate to validate adding data to the generator
    self._last_candidate = candidate

    return candidate

model_dump(*args, **kwargs)

overwrite model dump to remove faux class attrs

Source code in xopt/generator.py 
152
153
154
155
156
157
158
159
160
def model_dump(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
    """overwrite model dump to remove faux class attrs"""

    res = super().model_dump(*args, **kwargs)

    res.pop("supports_batch_generation", None)
    res.pop("supports_multi_objective", None)

    return res

reset()

Reset the generator.

Source code in xopt/generators/sequential/sequential_generator.py 
167
168
169
170
171
172
173
def reset(self):
    """
    Reset the generator.
    """
    self.is_active = False
    self._last_candidate = None
    self._reset()

set_data(data)

Set the full dataset for the generator. Typically only used when loading from a save file. This skips active generator lockout.

Parameters:

Name Type Description Default
data DataFrame

The data to set.

 required
Source code in xopt/generators/sequential/sequential_generator.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def set_data(self, data: pd.DataFrame):
    """
    Set the full dataset for the generator. Typically only used when loading from a save file. This skips active
    generator lockout.

    Parameters
    ----------
    data : pd.DataFrame
        The data to set.
    """
    # TODO: make a flag for generator that support multiple data sets
    if self._data_set:
        raise SeqGeneratorError(
            "Data has already been initialized for this generator."
        )
    self._set_data(data)
    self._data_set = True

validate_point(point)

determine if an input point was generated by the generator

Source code in xopt/generators/sequential/sequential_generator.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
def validate_point(self, point: Dict[str, float]):
    """determine if an input point was generated by the generator"""
    last_candidate = np.array(
        [self._last_candidate[0][ele] for ele in self.vocs.variable_names]
    )
    point_variables = np.array(
        [point[ele] for ele in self.vocs.variable_names]
    ).flatten()
    if not np.allclose(last_candidate, point_variables, atol=0.0, rtol=1e-6):
        raise SeqGeneratorError(
            "Cannot add data that was not generated by the generator when generator is active. "
            "Call reset() to reset the generator first in order to add data via other methods."
        )

yaml(**kwargs)

serialize first then dump to yaml string

Source code in xopt/pydantic.py 
231
232
233
234
235
236
237
238
def yaml(self, **kwargs):
    """serialize first then dump to yaml string"""
    output = json.loads(
        self.to_json(
            **kwargs,
        )
    )
    return yaml.dump(output)