Source code for hypertunity.reports.tensorboard

import os
import sys
from typing import Any, Dict, List, Union

import tinydb

from hypertunity import utils
from hypertunity.domain import Domain, Sample
from hypertunity.optimisation.base import HistoryPoint

from .base import Reporter

try:
    import tensorflow as tf
    from tensorboard.plugins.hparams import api as hp
except ImportError as err:
    raise ImportError("Install tensorflow>=1.14 and tensorboard>=1.14 "
                      "to support the HParams plugin.") from err


__all__ = [
    "Tensorboard"
]

EAGER_MODE = tf.executing_eagerly()
session_builder = tf.compat.v1.Session
if str(tf.version.VERSION) < "2.":
    summary_file_writer = tf.compat.v2.summary.create_file_writer
    summary_scalar = tf.compat.v2.summary.scalar
else:
    summary_file_writer = tf.summary.create_file_writer
    summary_scalar = tf.summary.scalar


[docs]class Tensorboard(Reporter): """A :class:`Reporter` subclass to visualise the results in Tensorboard. It utilises Tensorboard's HParams plugin as a dashboard for the summary of the optimisation. This class prepares and creates entries with the scalar data of the experiment trials, containing the domain sample and the corresponding metrics. Notes: The user is responsible for launching TensorBoard in the browser. """
[docs] def __init__(self, domain: Domain, metrics: List[str], logdir: str, primary_metric: str = "", database_path: str = None): """Initialise the TensorBoard reporter. Args: domain: :class:`Domain`. The domain to which all evaluated samples belong. metrics: :obj:`List[str]`. The names of the metrics. logdir: :obj:`str`. Path to a folder for storing the Tensorboard events. primary_metric: (optional) :obj:`str`. Primary metric from `metrics`. This is used by the :py:meth:`format` method to determine the sorting column and the best value. Default is the first one. database_path: (optional) :obj:`str`. The path to the database for storing experiment history on disk. Default is in-memory storage. """ super(Tensorboard, self).__init__( domain, metrics, primary_metric, database_path ) self._hparams_domain = self._convert_to_hparams_domain(self.domain) if not os.path.exists(logdir): os.makedirs(logdir) self._logdir = logdir self._experiment_counter = 0 self._set_up() print(f"Run 'tensorboard --logdir={logdir}' to launch " f"the visualisation in TensorBoard", file=sys.stderr)
@staticmethod def _convert_to_hparams_domain(domain: Domain) -> Dict[str, hp.HParam]: hparams = {} for var_name, dim in domain.flatten().items(): dim_type = Domain.get_type(dim) joined_name = utils.join_strings(var_name, join_char="/") if dim_type == Domain.Continuous: hp_dim_type = hp.RealInterval vals = list(map(float, dim)) elif dim_type in [Domain.Discrete, Domain.Categorical]: hp_dim_type = hp.Discrete vals = (dim,) else: raise TypeError( f"Cannot map subdomain of type {dim_type} " f"to a known HParams domain." ) hparams[joined_name] = hp.HParam(joined_name, hp_dim_type(*vals)) return hparams def _convert_to_hparams_sample(self, sample: Sample) -> Dict[hp.HParam, Any]: hparams = {} for name, val in sample: joined_name = utils.join_strings(name, join_char="/") hparams[self._hparams_domain[joined_name]] = val return hparams def _set_up(self): with summary_file_writer(self._logdir).as_default(): hp.hparams_config( hparams=self._hparams_domain.values(), metrics=[hp.Metric(m) for m in self.metrics]) @staticmethod def _log_tf_eager_mode(params, metrics, full_experiment_dir): """Log in eager mode.""" with summary_file_writer(full_experiment_dir).as_default(): hp.hparams(params) for metric_name, metric_value in metrics.items(): summary_scalar(metric_name, metric_value.value, step=1) @staticmethod def _log_tf_graph_mode(params, metrics, full_experiment_dir): """Log in legacy graph execution mode with session creation.""" with summary_file_writer(full_experiment_dir).as_default() as fw, session_builder() as sess: sess.run(fw.init()) sess.run(hp.hparams(params)) for metric_name, metric_value in metrics.items(): sess.run(summary_scalar(metric_name, metric_value.value, step=1)) sess.run(fw.flush()) def _log_history_point(self, entry: HistoryPoint, experiment_dir: str = None): """Create an entry for a :class:`HistoryPoint` in Tensorboard. Args: entry: :class:`HistoryPoint`. The sample and evaluation metrics to log. experiment_dir: (optional) :obj:`str`. The directory name where to store all experiment related data. It will be prefixed by the `logdir` path which is provided on initialisation of the :class:`Tensorboard` object. Default is 'experiment_[number]'. """ converted = self._convert_to_hparams_sample(entry.sample) if not experiment_dir: experiment_dir = f"experiment_{str(self._experiment_counter)}" self._experiment_counter += 1 full_experiment_dir = os.path.join(self._logdir, experiment_dir) if EAGER_MODE: self._log_tf_eager_mode(converted, entry.metrics, full_experiment_dir) else: self._log_tf_graph_mode(converted, entry.metrics, full_experiment_dir)
[docs] def from_database(self, database: Union[str, tinydb.TinyDB], table: str = None): """Load history from a database supplied as a path to a file or a :obj:`tinydb.TinyDB` object. Args: database: :obj:`str` or :obj:`tinydb.TinyDB`. The database to load. table: (optional) :obj:`str`. The table to load from the database. This argument is not required if the database has only one table. Raises: :class:`ValueError`: if the database contains more than one table and `table` is not given. """ super(Tensorboard, self).from_database(database, table) for doc in self._db_default_table: history_point = self._convert_doc_to_history(doc) self._log_history_point(history_point)