Source code for deephyper.stopper._stopper

import abc
import copy

from numbers import Number

[docs]class Stopper(abc.ABC): """An abstract class describing the interface of a Stopper. Args: max_steps (int): the maximum number of calls to ``observe(budget, objective)``. """ def __init__(self, max_steps: int) -> None: assert max_steps > 0 self.max_steps = max_steps self.job = None # Initialize list to collect observations self.observed_budgets = [] self.observed_objectives = [] self._stop_was_called = False
[docs] def to_json(self): """Returns a dict version of the stopper which can be saved as JSON.""" json_format = type(self).__name__ return json_format
[docs] def transform_objective(self, objective: float): """Replaces the currently observed objective by the maximum objective observed from the start. Identity transformation by default.""" # prev_objective = ( # self.observed_objectives[-1] if len(self.observed_objectives) > 0 else None # ) # if prev_objective is not None: # objective = max(prev_objective, objective) return objective
@property def step(self): """Last observed step.""" return self.observed_budgets[-1]
[docs] def observe(self, budget: float, objective: float) -> None: """Observe a new objective value. Args: budget (float): the budget used to obtain the objective (e.g., the number of epochs). objective (float): the objective value to observe (e.g, the accuracy). """ objective = self.transform_objective(objective) self.observed_budgets.append(budget) self.observed_objectives.append(objective)
[docs] def stop(self) -> bool: """Returns ``True`` if the evaluation should be stopped and ``False`` otherwise. Returns: bool: ``(step >= max_steps)``. """ if not (self._stop_was_called): self._stop_was_called = True, "_completed", False) if not (isinstance(self.observations[-1][-1], Number)): return True if self.step >= self.max_steps:, "_completed", True) return True return False
@property def observations(self) -> list: """Returns a copy of the list of observations with 0-index the budgets and 1-index the objectives.""" obs = [self.observed_budgets, self.observed_objectives] return copy.deepcopy(obs) @property def objective(self): """Last observed objective.""" return self.observations[-1][-1]