Scipy Minimize Generator

ScipyGenerator exposes scipy's optimize.minimize methods through Xopt's sequential ingest/suggest interface.

Integration Model

Xopt evaluates objective functions externally, one point at a time. scipy.optimize.minimize expects an in-process callable objective. ScipyGenerator bridges this mismatch by running one persistent scipy session in a worker thread:

1. A cache is built from prior evaluations (data) keyed by rounded variable values.
2. A single minimize call starts in a background thread.
3. The objective callback first checks the cache.
4. For an uncached point, that point is pushed to Xopt via a request queue.
5. Xopt evaluates the point externally and calls add_data.
6. The objective value is sent back through a response queue, and the same scipy run continues from in-memory state.

Only the first add_data call answering a pending request sends a response to scipy. Repeated submissions of the same candidate remain in the data history, but cannot answer a later request. Values are also cached under the requested coordinates, so small readback differences accepted by sequential point validation do not cause cache misses after a session restart.

Performance Notes

• Cache reconstruction is O(N) when a session starts or data is reloaded.
• The active scipy run is maintained between step calls; minimize is not restarted each step.
• Keys are rounded to 12 decimals before cache lookup to reduce floating-point key mismatch issues.

Supported Methods

method is validated against bounded scipy methods supported by this wrapper:

• Nelder-Mead
• Powell
• L-BFGS-B
• TNC
• SLSQP
• trust-constr
• COBYLA
• COBYQA

Invalid or empty method names fail validation.

Session Lifecycle and Errors

• reset() stops the active worker session and clears the result and completion state.
• set_data(...) stops any active session, clears completion state, reloads data, and rebuilds the cache.
• Each worker retains its own stop event, queues, and cache. Shutdown waits up to two seconds and warns if the worker is still running in scipy; that worker cannot consume responses or publish requests, results, or errors into a subsequent session.
• result exposes scipy's OptimizeResult after termination, including success, message, and solver-specific fields such as nit. It is None before termination.
• is_done is true once scipy terminates, including unsuccessful termination (for example, reaching an iteration limit). Check result.success to distinguish successful convergence.
• Once terminated, later generate() calls do not restart scipy or rebuild the cache. For compatibility, they still return the latest known data row, with a RuntimeWarning emitted once per session on this fallback path. Further explicit Xopt.step() calls will re-evaluate and append this duplicate; stop stepping when generator.is_done is true.
• Common scipy ValueError messages are remapped to clearer runtime errors (for example unsupported solver availability or bound handling).

State Restoration

For model-level roundtrips, use pydantic serialization:

• model_dump()
• model_validate(...)

Then restore the evaluation history with set_data(...) so the cache and last outcome are reconstructed before continuing optimization.

The class also defines __getstate__ and __setstate__ for explicit state handling of non-picklable runtime thread objects. Pickling does not stop the original worker. Pickle and deep-copy roundtrips retain completed results and the completion latch without sharing runtime thread objects; model-level roundtrips restart from the restored evaluation history instead. Requested-coordinate cache entries are retained through reset(), pickle, and deep-copy roundtrips. Replacing the dataset with set_data(...) clears these entries; model-level roundtrips rebuild from observed coordinates in the restored history.

Configuration

Typical fields:

• method: scipy minimization method name, e.g. Powell, Nelder-Mead, L-BFGS-B.
• initial_point: optional starting point dictionary.
• tol, options: passed directly to scipy.optimize.minimize.
• scipy_kwargs: additional keyword arguments forwarded to scipy.

ScipyGenerator

Bases: SequentialGenerator

Sequential wrapper around scipy.optimize.minimize.

Integration model

Xopt uses ingest/suggest semantics (one new point per step), while scipy minimize expects direct access to objective evaluations. This class bridges the two by running one persistent minimize call in a worker thread:

1. Build a cache from existing Xopt observations.
2. Start scipy.optimize.minimize once with an objective function that first checks the cache.
3. If scipy asks for an unseen point, send that point to Xopt and block until Xopt provides the external objective value.
4. Continue the same scipy run from in-memory state until termination.

Performance implications

• _build_cache is O(N + R), where N counts history rows and R counts requested-coordinate cache entries. It runs when a session starts, data is replaced, or state is copied or serialized.
• The active scipy run is maintained in memory between step calls.
• Point keys are rounded (12 decimals) before cache lookup to avoid fragile floating-point equality checks.

Session lifecycle

Each worker retains its own stop event, queues, and cache; the cache lock is shared across sessions. A stopped worker may finish numeric work later, but cannot publish status or exchange requests/responses with a newer session. Each delivered worker request has at most one response sent back to scipy. A returned OptimizeResult is latched through result and is_done until reset or data replacement. Subsequent generation warns once and returns the last data row, or raises if no evaluated data is available.

Source code in xopt/generators/sequential/scipy.py

class ScipyGenerator(SequentialGenerator):
    """
    Sequential wrapper around ``scipy.optimize.minimize``.

    Integration model
    -----------------
    Xopt uses ingest/suggest semantics (one new point per ``step``), while scipy
    ``minimize`` expects direct access to objective evaluations. This class bridges
    the two by running one persistent ``minimize`` call in a worker thread:

    1. Build a cache from existing Xopt observations.
    2. Start ``scipy.optimize.minimize`` once with an objective function that first
         checks the cache.
    3. If scipy asks for an unseen point, send that point to Xopt and block until
         Xopt provides the external objective value.
    4. Continue the same scipy run from in-memory state until termination.

    Performance implications
    ------------------------
    - ``_build_cache`` is O(N + R), where N counts history rows and R counts
        requested-coordinate cache entries. It runs when a session starts,
        data is replaced, or state is copied or serialized.
    - The active scipy run is maintained in memory between ``step`` calls.
    - Point keys are rounded (12 decimals) before cache lookup to avoid fragile
        floating-point equality checks.

    Session lifecycle
    -----------------
    Each worker retains its own stop event, queues, and cache; the cache lock is
    shared across sessions. A stopped worker may finish numeric work later, but
    cannot publish status or exchange requests/responses with a newer session.
    Each delivered worker request has at most one response sent back to scipy.
    A returned OptimizeResult is latched through ``result`` and ``is_done`` until
    reset or data replacement. Subsequent generation warns once and returns the
    last data row, or raises if no evaluated data is available.

    """

    name = "scipy"
    supports_single_objective: bool = True

    method: str = Field(
        "Powell",
        description="Method name passed to scipy.optimize.minimize (e.g. 'Powell', 'Nelder-Mead').",
    )
    initial_point: Optional[Dict[str, float]] = None
    tol: Optional[float] = Field(
        1e-8, description="Termination tolerance passed to scipy.optimize.minimize"
    )
    options: Dict[str, Any] = Field(
        default_factory=dict,
        description="Options dictionary passed directly to scipy.optimize.minimize",
    )
    scipy_kwargs: Dict[str, Any] = Field(
        default_factory=dict,
        description="Additional keyword arguments passed to scipy.optimize.minimize.",
    )

    # Internal state
    _last_outcome: Optional[float] = None
    _cache: Dict[tuple, float] = PrivateAttr(default_factory=dict)
    # Preserve answers at requested coordinates, even if observed readbacks differ.
    _requested_cache: Dict[tuple, float] = PrivateAttr(default_factory=dict)
    # Rounded coordinates of the delivered candidate still awaiting its first answer.
    _pending_request: Optional[tuple] = PrivateAttr(default=None)
    _cache_lock: Lock = PrivateAttr(default_factory=Lock)
    _request_queue: Queue = PrivateAttr(default_factory=Queue)
    _response_queue: Queue = PrivateAttr(default_factory=Queue)
    _stop_event: Event = PrivateAttr(default_factory=Event)
    _session_thread: Optional[Thread] = PrivateAttr(default=None)
    _session_exception: Optional[Exception] = PrivateAttr(default=None)
    _session_finished: bool = PrivateAttr(default=False)
    _result: Optional[OptimizeResult] = PrivateAttr(default=None)
    _completion_warned: bool = PrivateAttr(default=False)
    _stop_token: object = PrivateAttr(default_factory=object)

    _runtime_private_attr_names = {
        "_cache_lock",
        "_request_queue",
        "_response_queue",
        "_stop_event",
        "_session_thread",
        "_session_exception",
        "_session_finished",
        "_stop_token",
    }

    @field_validator("method")
    def validate_method(cls, v: str):
        """Strip whitespace and require a method from the bounded-method allowlist."""
        value = v.strip()
        if value not in BOUNDED_METHODS:
            raise ValueError(
                f"scipy method '{value}' is not supported; choose one of {BOUNDED_METHODS}"
            )
        return value

    @field_validator("initial_point")
    def validate_initial_point(cls, v: Optional[Dict[str, float]]):
        """Ensure that ``initial_point`` is either omitted or contains coordinates."""
        if v is not None and len(v) == 0:
            raise ValueError("initial_point cannot be an empty dictionary")
        return v

    def __deepcopy__(self, memo):
        """Copy logical optimization state without copying a running worker.

        Private sequential state is absent from ``model_dump`` and must be
        restored explicitly so a copy can accept an outstanding candidate's
        evaluation. An unfinished optimization restarts lazily using its cache;
        a completed copy keeps its result and does not start another worker.
        """
        copied = self.__class__.model_validate(self.model_dump())
        memo[id(self)] = copied
        if self.data is not None:
            copied.data = self.data.copy(deep=True)
        copied._last_candidate = copy.deepcopy(self._last_candidate, memo)
        copied._data_set = self._data_set
        copied._last_outcome = self._last_outcome
        copied._requested_cache = copy.deepcopy(self._requested_cache, memo)
        copied._pending_request = self._pending_request
        copied._cache = copied._build_cache()
        copied._result = copy.deepcopy(self._result, memo)
        copied._completion_warned = self._completion_warned
        return copied

    def __getstate__(self):
        """Snapshot replayable state without stopping the original worker.

        Synchronization objects and worker status are process-local. Candidate,
        requested-coordinate cache, and completed-result state remain picklable
        so restoration can resume the ingest/suggest contract without sharing a worker.
        """
        state = super().__getstate__().copy()
        # Filtering a copied mapping must not remove the live object's primitives.
        private = dict(state.get("__pydantic_private__", {}) or {})

        for key in self._runtime_private_attr_names:
            private.pop(key, None)

        private["_cache"] = self._build_cache()
        state["__pydantic_private__"] = private
        return state

    def __setstate__(self, state):
        """Restore logical state with fresh, idle synchronization objects.

        Pending evaluations can still be ingested before generation resumes.
        No response is queued until a new worker has delivered a request.
        """
        super().__setstate__(state)
        self._cache_lock = Lock()
        self._request_queue = Queue()
        self._response_queue = Queue()
        self._stop_event = Event()
        self._session_thread = None
        self._session_exception = None
        self._session_finished = False
        self._stop_token = object()

        self._cache = self._build_cache()
        if self.data is not None and len(self.data) > 0:
            objective_data = get_objective_data(self.vocs, self.data).to_numpy()[:, 0]
            self._last_outcome = float(objective_data[-1])

    @property
    def result(self) -> Optional[OptimizeResult]:
        """Return scipy's termination result, or None before it is available.

        Inspect ``success`` and ``message`` to distinguish convergence from
        iteration limits or other solver-reported failures.
        """
        return self._result

    @property
    def is_done(self) -> bool:
        """Whether scipy returned an OptimizeResult, successful or otherwise.

        Worker exceptions and cooperative shutdown do not produce a result and
        therefore do not set this flag.
        """
        return self._result is not None

    @property
    def x0(self) -> np.ndarray:
        """Return the optimization start point from config or from the latest dataset row."""
        if self.initial_point is not None:
            missing = [
                k for k in self.vocs.variable_names if k not in self.initial_point
            ]
            if missing:
                raise ValueError(
                    f"initial_point is missing values for variables: {missing}"
                )
            return np.array(
                [self.initial_point[k] for k in self.vocs.variable_names], dtype=float
            )
        return self._get_initial_point()[0]

    def _reset(self):
        """Clear session/completion state while retaining history and replay values."""
        self._stop_session()
        self._last_outcome = None

    def _set_data(self, data: pd.DataFrame):
        """Replace history and discard request mappings belonging to the old dataset.

        Unlike reset, replacement cannot reuse requested-coordinate values:
        they may no longer correspond to observations in the supplied history.
        """
        self._stop_session()
        self.data = data
        self._requested_cache = {}
        if len(data) > 0:
            objective_data = get_objective_data(self.vocs, data).to_numpy()[:, 0]
            self._last_outcome = float(objective_data[-1])
        self._cache = self._build_cache()

    def _add_data(self, new_data: pd.DataFrame):
        """Cache an evaluation and answer a delivered request at most once.

        Sequential point validation permits small readback/setpoint differences.
        Cache the first answer under the requested coordinates as well as the
        observed coordinates so replay reproduces the values scipy received.
        Submissions without a pending request remain in history but do not queue
        responses that could be consumed by a later request.
        """
        if len(new_data) == 0:
            return
        objective_data = get_objective_data(self.vocs, new_data).to_numpy()[:, 0]
        self._last_outcome = float(objective_data[-1])

        x_value = get_variable_data(self.vocs, new_data).to_numpy(dtype=float)[-1]
        with self._cache_lock:
            observed_key = self._point_key(x_value)
            # Protect requested-coordinate answers from later observed-history updates.
            self._cache[observed_key] = self._requested_cache.get(
                observed_key, self._last_outcome
            )
            if self._pending_request is not None:
                self._requested_cache[self._pending_request] = self._last_outcome
                self._cache[self._pending_request] = self._last_outcome
                # Consume the pending request before waking the worker.
                self._pending_request = None
                if self._session_thread is not None and self._session_thread.is_alive():
                    self._response_queue.put(self._last_outcome)

    def _point_key(self, x: np.ndarray, decimals: int = 12) -> tuple:
        """Convert a floating-point vector to a rounded hashable cache key."""
        return tuple(np.round(np.array(x, dtype=float), decimals=decimals))

    def _build_cache(self) -> dict[tuple, float]:
        """Build an O(N + R) replay cache from N history rows and R requested answers.

        Requested-point answers take precedence over observed history because
        duplicates and accepted readback differences must not change replay.
        """
        if self.data is None or len(self.data) == 0:
            return {}

        # Build a deterministic replay table from prior Xopt observations.
        x_data = get_variable_data(self.vocs, self.data).to_numpy(dtype=float)
        y_data = get_objective_data(self.vocs, self.data).to_numpy()[:, 0]

        cache = {self._point_key(x): float(y) for x, y in zip(x_data, y_data)}
        cache.update(self._requested_cache)
        return cache

    def _map_value_error(self, ex: ValueError) -> Optional[RuntimeError]:
        """Map scipy ``ValueError`` messages to clearer runtime configuration errors."""
        msg = str(ex).lower()
        if "unknown solver" in msg:
            return RuntimeError(
                f"scipy method '{self.method}' is not available in this environment."
            )
        if "cannot handle bounds" in msg:
            return RuntimeError(
                f"scipy method '{self.method}' does not support bounds; choose a bounded method (e.g. 'Powell', 'L-BFGS-B', 'TNC', 'SLSQP', 'trust-constr')."
            )
        return None

    def _raise_session_error_if_present(self):
        """Raise any worker exception in the caller thread, with friendly remapping."""
        if self._session_exception is None:
            return

        ex = self._session_exception
        self._session_exception = None
        if isinstance(ex, ValueError):
            mapped = self._map_value_error(ex)
            if mapped is not None:
                raise mapped from ex
        raise ex

    def _objective(
        self,
        x: np.ndarray,
        *,
        stop_event: Event,
        request_queue: Queue,
        response_queue: Queue,
        cache: Dict[tuple, float],
        cache_lock: Lock,
        stop_token: object,
    ) -> float:
        """Objective callback used by scipy.

        This method first serves values from cache. For uncached points, it sends
        the requested point to the main thread and blocks until the corresponding
        externally evaluated objective value is provided.

        All runtime arguments belong to this worker's session, not necessarily
        the generator's current session. Never resolve queues, cache, or stop
        signals through ``self`` here: shutdown may have replaced them already.
        """
        if stop_event.is_set():
            raise _StopSession()

        key = self._point_key(x)
        with cache_lock:
            if key in cache:
                return cache[key]

        request_queue.put(np.array(x, dtype=float))
        # Do not hold the cache lock while waiting; add_data needs it to answer.
        response = response_queue.get()

        # The sentinel wakes a blocked callback; the event rejects answers after stop.
        if response is stop_token or stop_event.is_set():
            raise _StopSession()

        y_value = float(response)
        with cache_lock:
            cache[key] = y_value
        return y_value

    def _run_session(
        self,
        stop_event: Event,
        request_queue: Queue,
        response_queue: Queue,
        cache: Dict[tuple, float],
        cache_lock: Lock,
        stop_token: object,
    ):
        """Run scipy using session-bound resources and publish only current status.

        Stop-event identity acts as a session token. Publication is guarded by
        the same lock used to signal shutdown, so an abandoned worker cannot
        overwrite a newer session's result, exception, or finished flag.
        """
        mins, maxs = np.array(self.vocs.bounds).T
        bounds = list(zip(mins, maxs))

        minimize_kwargs = {
            "method": self.method,
            "bounds": bounds,
            "tol": self.tol,
            "options": self.options,
            **self.scipy_kwargs,
        }

        objective = partial(
            self._objective,
            stop_event=stop_event,
            request_queue=request_queue,
            response_queue=response_queue,
            cache=cache,
            cache_lock=cache_lock,
            stop_token=stop_token,
        )
        try:
            result = minimize(objective, self.x0, **minimize_kwargs)
            with cache_lock:
                if self._stop_event is stop_event and not stop_event.is_set():
                    self._result = result
        except _StopSession:
            # Cooperative shutdown is not an optimization failure.
            pass
        except Exception as ex:
            with cache_lock:
                if self._stop_event is stop_event and not stop_event.is_set():
                    self._session_exception = ex
        finally:
            with cache_lock:
                if self._stop_event is stop_event and not stop_event.is_set():
                    self._session_finished = True

    def _start_session_if_needed(self):
        """Start an unfinished optimization lazily, replaying previously cached values.

        A retained result latches termination, preventing expensive trajectory
        replay on every generate call after scipy has already finished.
        """
        if self.is_done:
            return
        if self._session_thread is not None and self._session_thread.is_alive():
            return

        self._stop_event = Event()
        self._session_exception = None
        self._session_finished = False
        self._request_queue = Queue()
        self._response_queue = Queue()
        self._cache = self._build_cache()

        self._session_thread = Thread(
            target=self._run_session,
            args=(
                self._stop_event,
                self._request_queue,
                self._response_queue,
                self._cache,
                self._cache_lock,
                self._stop_token,
            ),
            daemon=True,
        )
        self._session_thread.start()

    def _stop_session(self):
        """Signal shutdown, wait briefly, and detach the session's runtime state.

        Python cannot interrupt scipy numeric work in another thread. If the
        bounded join expires, warn and leave the old worker's event set; its
        session-bound resources and guarded status updates isolate it from the
        replacement session. Evaluation history and requested answers survive.
        A worker still in numeric code notices the stop on its next objective
        callback, or has its eventual result/exception discarded.
        """
        with self._cache_lock:
            self._stop_event.set()
        if self._session_thread is not None and self._session_thread.is_alive():
            self._response_queue.put(self._stop_token)
            # Joining under the cache lock would block worker shutdown/publication.
            self._session_thread.join(timeout=2.0)
            if self._session_thread.is_alive():
                warnings.warn(
                    "scipy worker did not stop within 2 seconds; "
                    "it remains isolated from subsequent sessions.",
                    RuntimeWarning,
                    stacklevel=2,
                )

        self._session_thread = None
        self._session_exception = None
        self._session_finished = False
        self._result = None
        self._completion_warned = False
        self._pending_request = None
        self._request_queue = Queue()
        self._response_queue = Queue()
        self._stop_event = Event()

    def _generate(self, first_gen: bool = False) -> Optional[List[Dict[str, float]]]:
        """Return the next candidate requested by the live scipy session.

        The ``first_gen`` argument is accepted to satisfy the sequential generator
        interface; new candidates come from the active persistent session.

        Delivery marks a request pending so add_data can answer it once. Queue
        polling also checks worker failure/termination rather than blocking
        forever when scipy exits without requesting another evaluation.
        After termination, retain the historical last-row fallback and warn once:
        explicit Xopt.step calls will still evaluate and append that duplicate.
        """
        self._start_session_if_needed()

        while not self.is_done:
            self._raise_session_error_if_present()

            try:
                requested_x = self._request_queue.get(timeout=0.05)
                self._pending_request = self._point_key(requested_x)
                inputs = dict(zip(self.vocs.variable_names, requested_x.tolist()))
                if self.vocs.constants is not None:
                    inputs.update(self.vocs.constants)
                return [inputs]
            except Empty:
                if self._session_finished:
                    self._raise_session_error_if_present()
                    break

        # Preserve the historical fallback without replaying a completed optimization.
        if self.data is None or len(self.data) == 0:
            raise RuntimeError("scipy minimize converged without available data")

        result = self.result
        if result is None:
            raise RuntimeError("scipy minimize session stopped without a result")

        if not self._completion_warned:
            warnings.warn(
                f"scipy minimize has terminated: {result.message}. "
                "Returning the latest data point; further Xopt.step() calls will "
                "re-evaluate it. Check is_done and result, or call reset() to restart.",
                RuntimeWarning,
                stacklevel=2,
            )
            self._completion_warned = True

        inputs = self.data[self.vocs.variable_names].iloc[-1].to_dict()
        if self.vocs.constants is not None:
            inputs.update(self.vocs.constants)
        return [inputs]

is_done property

Whether scipy returned an OptimizeResult, successful or otherwise.

Worker exceptions and cooperative shutdown do not produce a result and therefore do not set this flag.

result property

Return scipy's termination result, or None before it is available.

Inspect success and message to distinguish convergence from iteration limits or other solver-reported failures.

x0 property

Return the optimization start point from config or from the latest dataset row.

__init__(**kwargs)

Initialize the generator.

Source code in xopt/generator.py

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

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

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

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

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

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

Ensure that initial_point is either omitted or contains coordinates.

Source code in xopt/generators/sequential/scipy.py

@field_validator("initial_point")
def validate_initial_point(cls, v: Optional[Dict[str, float]]):
    """Ensure that ``initial_point`` is either omitted or contains coordinates."""
    if v is not None and len(v) == 0:
        raise ValueError("initial_point cannot be an empty dictionary")
    return v

validate_method(v)

Strip whitespace and require a method from the bounded-method allowlist.

Source code in xopt/generators/sequential/scipy.py

@field_validator("method")
def validate_method(cls, v: str):
    """Strip whitespace and require a method from the bounded-method allowlist."""
    value = v.strip()
    if value not in BOUNDED_METHODS:
        raise ValueError(
            f"scipy method '{value}' is not supported; choose one of {BOUNDED_METHODS}"
        )
    return value

validate_point(point)

determine if an input point was generated by the generator

Source code in xopt/generators/sequential/sequential_generator.py

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

def yaml(self, **kwargs: Any) -> str:
    """serialize first then dump to yaml string"""
    output = json.loads(
        self.to_json(
            **kwargs,
        )
    )
    return yaml.dump(output)