Source code for orion.benchmark

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Benchmark definition
======================
"""
import copy
import itertools

from tabulate import tabulate

from orion.client import create_experiment


[docs]class Benchmark: """ Benchmark definition Parameters ---------- name: str Name of the benchmark algorithms: list, optional Algorithms used for benchmark, each algorithm can be a string or dict. targets: list, optional Targets for the benchmark, each target will be a dict with two keys. assess: list Assessment objects task: list Task objects """ def __init__(self, name, algorithms, targets): self._id = None self.name = name self.algorithms = algorithms self.targets = targets self.metadata = {} self.studies = []
[docs] def setup_studies(self): """Setup studies to run for the benchmark. Benchmark `algorithms`, together with each `task` and `assessment` combination define a study. """ for target in self.targets: assessments = target["assess"] tasks = target["task"] for assess, task in itertools.product(*[assessments, tasks]): study = Study(self, self.algorithms, assess, task) study.setup_experiments() self.studies.append(study)
[docs] def process(self): """Run studies experiment""" for study in self.studies: study.execute()
[docs] def status(self, silent=True): """Display benchmark status""" total_exp_num = 0 complete_exp_num = 0 total_trials = 0 benchmark_status = [] for study in self.studies: for status in study.status(): column = dict() column["Algorithms"] = status["algorithm"] column["Assessments"] = status["assessment"] column["Tasks"] = status["task"] column["Total Experiments"] = status["experiments"] total_exp_num += status["experiments"] column["Completed Experiments"] = status["completed"] complete_exp_num += status["completed"] column["Submitted Trials"] = status["trials"] total_trials += status["trials"] benchmark_status.append(column) if not silent: print( "Benchmark name: {}, Experiments: {}/{}, Submitted trials: {}".format( self.name, complete_exp_num, total_exp_num, total_trials ) ) self._pretty_table(benchmark_status) return benchmark_status
[docs] def analysis(self): """Return all the assessment figures""" figures = [] for study in self.studies: figure = study.analysis() figures.append(figure) return figures
[docs] def experiments(self, silent=True): """Return all the experiments submitted in benchmark""" experiment_table = [] for study in self.studies: for exp in study.experiments(): exp_column = dict() stats = exp.stats exp_column["Algorithm"] = list(exp.configuration["algorithms"].keys())[ 0 ] exp_column["Experiment Name"] = exp.name exp_column["Number Trial"] = len(exp.fetch_trials()) exp_column["Best Evaluation"] = stats["best_evaluation"] experiment_table.append(exp_column) if not silent: print("Total Experiments: {}".format(len(experiment_table))) self._pretty_table(experiment_table) return experiment_table
def _pretty_table(self, dict_list): """ Print a list of same format dict as pretty table with IPython disaply(notebook) or tablute :param dict_list: a list of dict where each dict has the same keys. """ try: from IPython.display import HTML, display display( HTML( tabulate( dict_list, headers="keys", tablefmt="html", stralign="center", numalign="center", ) ) ) except ImportError: table = tabulate( dict_list, headers="keys", tablefmt="grid", stralign="center", numalign="center", ) print(table) # pylint: disable=invalid-name @property def id(self): """Id of the benchmark in the database if configured. Value is `None` if the benchmark is not configured. """ return self._id @property def configuration(self): """Return a copy of an `Benchmark` configuration as a dictionary.""" config = dict() config["name"] = self.name config["algorithms"] = self.algorithms targets = [] for target in self.targets: str_target = {} assessments = target["assess"] str_assessments = dict() for assessment in assessments: str_assessments.update(assessment.configuration) str_target["assess"] = str_assessments tasks = target["task"] str_tasks = dict() for task in tasks: str_tasks.update(task.configuration) str_target["task"] = str_tasks targets.append(str_target) config["targets"] = targets if self.id is not None: config["_id"] = self.id return copy.deepcopy(config)
[docs]class Study: """ A study is one assessment and task combination in the `Benchmark` targets. It will build and run experiments for all the algorithms for that task. Parameters ---------- benchmark: `Benchmark` instance algorithms: list Algorithms used for benchmark, each algorithm can be a string or dict. assessment: list `Assessment` instance task: list `Task` instance """ def __init__(self, benchmark, algorithms, assessment, task): self.algorithms = algorithms self.assessment = assessment self.task = task self.benchmark = benchmark self.assess_name = type(self.assessment).__name__ self.task_name = type(self.task).__name__ self.experiments_info = []
[docs] def setup_experiments(self): """Setup experiments to run of the study""" max_trials = self.task.max_trials task_num = self.assessment.task_num space = self.task.get_search_space() for task_index in range(task_num): for algo_index, algorithm in enumerate(self.algorithms): experiment_name = ( self.benchmark.name + "_" + self.assess_name + "_" + self.task_name + "_" + str(task_index) + "_" + str(algo_index) ) experiment = create_experiment( experiment_name, space=space, algorithms=algorithm, max_trials=max_trials, ) self.experiments_info.append((task_index, experiment))
[docs] def execute(self): """Execute all the experiments of the study""" max_trials = self.task.max_trials for _, experiment in self.experiments_info: # TODO: it is a blocking call experiment.workon(self.task, max_trials)
[docs] def status(self): """Return status of the study""" algorithm_tasks = {} for _, experiment in self.experiments_info: trials = experiment.fetch_trials() algorithm_name = list(experiment.configuration["algorithms"].keys())[0] if algorithm_tasks.get(algorithm_name, None) is None: task_state = { "algorithm": algorithm_name, "experiments": 0, "assessment": self.assess_name, "task": self.task_name, "completed": 0, "trials": 0, } else: task_state = algorithm_tasks[algorithm_name] task_state["experiments"] += 1 task_state["trials"] += len(trials) if experiment.is_done: task_state["completed"] += 1 algorithm_tasks[algorithm_name] = task_state return list(algorithm_tasks.values())
[docs] def analysis(self): """Return assessment figure""" return self.assessment.analysis(self.task_name, self.experiments_info)
[docs] def experiments(self): """Return all the experiments of the study""" exps = [] for _, experiment in self.experiments_info: exps.append(experiment) return exps
def __repr__(self): """Represent the object as a string.""" algorithms_list = list() for algorithm in self.algorithms: if isinstance(algorithm, dict): algorithm_name = list(algorithm.keys())[0] else: algorithm_name = algorithm algorithms_list.append(algorithm_name) return "Study(assessment=%s, task=%s, algorithms=[%s])" % ( self.assess_name, self.task_name, ",".join(algorithms_list), )