Source code for deephyper.evaluator._job

import copy
from import MutableMapping

from typing import Hashable

from import Storage, MemoryStorage
from deephyper.evaluator._run_function_utils import standardize_run_function_output
from deephyper.stopper._stopper import Stopper

[docs]class Job: """Represents an evaluation executed by the ``Evaluator`` class. Args: id (Any): unique identifier of the job. Usually an integer. config (dict): argument dictionnary of the ``run_function``. run_function (callable): function executed by the ``Evaluator`` """ # Job status states. READY = 0 RUNNING = 1 DONE = 2 def __init__(self, id, config: dict, run_function): = id self.rank = None self.config = copy.deepcopy(config) self.run_function = run_function self.status = self.READY self.output = { "objective": None, "metadata": {"timestamp_submit": None, "timestamp_gather": None}, } self.observations = None def __repr__(self) -> str: if self.rank is not None: return f"Job(id={}, rank={self.rank}, status={self.status}, config={self.config})" else: return f"Job(id={}, status={self.status}, config={self.config})" def __getitem__(self, index): cfg = copy.deepcopy(self.config) return (cfg, self.objective)[index] @property def parameters(self): return self.config @property def result(self): """Alias for the objective property.""" return self.objective @property def objective(self): """Objective returned by the run-function.""" return self.output["objective"] @property def metadata(self): """Metadata of the job stored in the output of run-function.""" return self.output["metadata"] def set_output(self, output): output = standardize_run_function_output(output) self.output["objective"] = output["objective"] self.output["metadata"].update(output["metadata"]) self.observations = output.get("observations", None) def create_running_job(self, storage, stopper): stopper = copy.deepcopy(stopper) rjob = RunningJob(, self.config, storage, stopper) if stopper is not None and hasattr(stopper, "job"): stopper.job = rjob return rjob
[docs]class RunningJob(MutableMapping): """A RunningJob is adapted Job object that is passed to the run-function as input. Args: id (Hashable, optional): The identifier of the job in the Storage. Defaults to None. parameters (dict, optional): The dictionnary of hyperparameters suggested. Defaults to None. storage (Storage, optional): The storage client used for the search. Defaults to None. stopper (Stopper, optional): The stopper object used for the evaluation. Defaults to None. """ def __init__( self, id: Hashable = None, parameters: dict = None, storage: Storage = None, stopper: Stopper = None, ) -> None: = id self.parameters = parameters if storage is None: = MemoryStorage() search_id = = else: = storage self.stopper = stopper self.obs = None def __getitem__(self, key): if key == "job_id": return int(".")[-1]) return self.parameters[key] def __setitem__(self, key, value): if key == "job_id": raise KeyError("Cannot change the 'job_id' of a running job.") self.parameters[key] = value def __delitem__(self, key): del self.parameters[key] def __iter__(self): return iter(self.parameters) def __len__(self): return len(self.parameters)
[docs] def record(self, budget: float, objective: float): """Records the current ``budget`` and ``objective`` values in the object and pass it to the stopper if one is being used. Args: budget (float): the budget used. objective (float): the objective value obtained. """ if self.stopper: self.stopper.observe(budget, objective) else: self.obs = objective
[docs] def stopped(self) -> bool: """Returns True if the RunningJob is using a Stopper and it is stopped. Otherwise it will return False.""" if self.stopper: return self.stopper.stop() else: return False
@property def objective(self): """If the RunningJob is using a Stopper then it will return observations from the it. Otherwise it will simply return the last objective value recorded.""" if self.stopper: return self.stopper.objective else: return self.obs