[Tune] Post-Experiment Tools (#4351)

python/ray/tune/ParallelCoordinatesVisualization.ipynb

@@ -1,129 +0,0 @@
-{
- "cells": [
-  {
-   "cell_type": "markdown",
-   "metadata": {},
-   "source": [
-    "# Tune Visualization"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "metadata": {},
-   "source": [
-    "In order to visualize results, please install `plotly` with the following command:\n",
-    "\n",
-    "  `pip install plotly`"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {
-    "collapsed": true
-   },
-   "outputs": [],
-   "source": [
-    "import os\n",
-    "import pandas as pd\n",
-    "from ray.tune.visual_utils import load_results_to_df, generate_plotly_dim_dict\n",
-    "import plotly\n",
-    "import plotly.graph_objs as go\n",
-    "plotly.offline.init_notebook_mode(connected=True)"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "metadata": {},
-   "source": [
-    "### Specify the directory where all your results are in the variable `RESULTS_DIR`."
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {
-    "collapsed": true
-   },
-   "outputs": [],
-   "source": [
-    "RESULTS_DIR = os.path.expanduser(\"~/ray_results\")\n",
-    "df = load_results_to_df(RESULTS_DIR)\n",
-    "[key for key in df]"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "metadata": {},
-   "source": [
-    "### Choose the fields you wish to visualize over in `GOOD_FIELDS`."
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {
-    "collapsed": true,
-    "scrolled": true
-   },
-   "outputs": [],
-   "source": [
-    "GOOD_FIELDS = ['experiment_id',\n",
-    "               'num_sgd_iter',\n",
-    "               'episode_len_mean',\n",
-    "               'episode_reward_mean']\n",
-    "\n",
-    "visualization_df = df[GOOD_FIELDS]\n",
-    "visualization_df = visualization_df.dropna()\n",
-    "visualization_df"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "metadata": {},
-   "source": [
-    "### Enjoy.\n",
-    "\n",
-    "Documentation for this Plotly visualization can be found here: https://plot.ly/python/parallel-coordinates-plot/"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {
-    "collapsed": true
-   },
-   "outputs": [],
-   "source": [
-    "data = [go.Parcoords(\n",
-    "            line = dict(color = 'blue'),\n",
-    "            dimensions = [generate_plotly_dim_dict(visualization_df, field) \n",
-    "                              for field in visualization_df])\n",
-    "]\n",
-    "\n",
-    "plotly.offline.iplot(data)"
-   ]
-  }
- ],
- "metadata": {
-  "kernelspec": {
-   "display_name": "Python 3",
-   "language": "python",
-   "name": "python3"
-  },
-  "language_info": {
-   "codemirror_mode": {
-    "name": "ipython",
-    "version": 3
-   },
-   "file_extension": ".py",
-   "mimetype": "text/x-python",
-   "name": "python",
-   "nbconvert_exporter": "python",
-   "pygments_lexer": "ipython3",
-   "version": "3.6.2"
-  }
- },
- "nbformat": 4,
- "nbformat_minor": 2
-}

python/ray/tune/analysis/__init__.py

@@ -0,0 +1,7 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+from ray.tune.analysis.experiment_analysis import ExperimentAnalysis
+
+__all__ = ["ExperimentAnalysis"]

python/ray/tune/analysis/experiment_analysis.py

@@ -0,0 +1,108 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import copy
+import glob
+import json
+import logging
+import os
+import pandas as pd
+
+from ray.tune.error import TuneError
+from ray.tune.util import flatten_dict
+
+logger = logging.getLogger(__name__)
+
+UNNEST_KEYS = ("config", "last_result")
+
+
+def unnest_checkpoints(checkpoints):
+    checkpoint_dicts = []
+    for g in checkpoints:
+        checkpoint = copy.deepcopy(g)
+        for key in UNNEST_KEYS:
+            if key not in checkpoint:
+                continue
+            try:
+                unnest_dict = flatten_dict(checkpoint.pop(key))
+                checkpoint.update(unnest_dict)
+            except Exception:
+                logger.debug("Failed to flatten dict.")
+        checkpoint = flatten_dict(checkpoint)
+        checkpoint_dicts.append(checkpoint)
+    return checkpoint_dicts
+
+
+class ExperimentAnalysis(object):
+    """Analyze results from a Tune experiment.
+
+    Parameters:
+        experiment_path (str): Path to where experiment is located.
+            Corresponds to Experiment.local_dir/Experiment.name
+
+    Example:
+        >>> tune.run(my_trainable, name="my_exp", local_dir="~/tune_results")
+        >>> analysis = ExperimentAnalysis(
+        >>>     experiment_path="~/tune_results/my_exp")
+    """
+
+    def __init__(self, experiment_path):
+        experiment_path = os.path.expanduser(experiment_path)
+        if not os.path.isdir(experiment_path):
+            raise TuneError(
+                "{} is not a valid directory.".format(experiment_path))
+        experiment_state_paths = glob.glob(
+            os.path.join(experiment_path, "experiment_state*.json"))
+        if not experiment_state_paths:
+            raise TuneError("No experiment state found!")
+        experiment_filename = max(
+            list(experiment_state_paths))  # if more than one, pick latest
+        with open(os.path.join(experiment_path, experiment_filename)) as f:
+            self._experiment_state = json.load(f)
+
+        if "checkpoints" not in self._experiment_state:
+            raise TuneError("Experiment state invalid; no checkpoints found.")
+        self._checkpoints = self._experiment_state["checkpoints"]
+        self._scrubbed_checkpoints = unnest_checkpoints(self._checkpoints)
+
+    def dataframe(self):
+        """Returns a pandas.DataFrame object constructed from the trials."""
+        return pd.DataFrame(self._scrubbed_checkpoints)
+
+    def stats(self):
+        """Returns a dictionary of the statistics of the experiment."""
+        return self._experiment_state.get("stats")
+
+    def runner_data(self):
+        """Returns a dictionary of the TrialRunner data."""
+        return self._experiment_state.get("runner_data")
+
+    def trial_dataframe(self, trial_id):
+        """Returns a pandas.DataFrame constructed from one trial."""
+        for checkpoint in self._checkpoints:
+            if checkpoint["trial_id"] == trial_id:
+                logdir = checkpoint["logdir"]
+                progress = max(glob.glob(os.path.join(logdir, "progress.csv")))
+                return pd.read_csv(progress)
+        raise ValueError("Trial id {} not found".format(trial_id))
+
+    def get_best_trainable(self, metric, trainable_cls):
+        """Returns the best Trainable based on the experiment metric."""
+        return trainable_cls(config=self.get_best_config(metric))
+
+    def get_best_config(self, metric):
+        """Retrieve the best config from the best trial."""
+        return self._get_best_trial(metric)["config"]
+
+    def _get_best_trial(self, metric):
+        """Retrieve the best trial based on the experiment metric."""
+        return max(
+            self._checkpoints, key=lambda d: d["last_result"].get(metric, 0))
+
+    def _get_sorted_trials(self, metric):
+        """Retrive trials in sorted order based on the experiment metric."""
+        return sorted(
+            self._checkpoints,
+            key=lambda d: d["last_result"].get(metric, 0),
+            reverse=True)

python/ray/tune/commands.py

@@ -2,8 +2,6 @@ from __future__ import absolute_import
 from __future__ import division
 from __future__ import print_function
 
-import glob
-import json
 import logging
 import os
 import sys
@@ -13,9 +11,10 @@ from datetime import datetime
 
 import pandas as pd
 from pandas.api.types import is_string_dtype, is_numeric_dtype
-from ray.tune.util import flatten_dict
 from ray.tune.result import TRAINING_ITERATION, MEAN_ACCURACY, MEAN_LOSS
 from ray.tune.trial import Trial
+from ray.tune.analysis import ExperimentAnalysis
+from ray.tune import TuneError
 try:
     from tabulate import tabulate
 except ImportError:
@@ -40,8 +39,6 @@ DEFAULT_PROJECT_INFO_KEYS = (
     "last_updated",
 )
 
-UNNEST_KEYS = ("config", "last_result")
-
 try:
     TERM_HEIGHT, TERM_WIDTH = subprocess.check_output(["stty", "size"]).split()
     TERM_HEIGHT, TERM_WIDTH = int(TERM_HEIGHT), int(TERM_WIDTH)
@@ -104,23 +101,6 @@ def print_format_output(dataframe):
     return table, dropped_cols, empty_cols
 
 
-def _get_experiment_state(experiment_path, exit_on_fail=False):
-    experiment_path = os.path.expanduser(experiment_path)
-    experiment_state_paths = glob.glob(
-        os.path.join(experiment_path, "experiment_state*.json"))
-    if not experiment_state_paths:
-        if exit_on_fail:
-            print("No experiment state found!")
-            sys.exit(0)
-        else:
-            return
-    experiment_filename = max(list(experiment_state_paths))
-
-    with open(experiment_filename) as f:
-        experiment_state = json.load(f)
-    return experiment_state
-
-
 def list_trials(experiment_path,
                 sort=None,
                 output=None,
@@ -142,22 +122,12 @@ def list_trials(experiment_path,
         desc (bool): Sort ascending vs. descending.
     """
     _check_tabulate()
-    experiment_state = _get_experiment_state(
-        experiment_path, exit_on_fail=True)
 
-    checkpoints = experiment_state["checkpoints"]
-
-    checkpoint_dicts = []
-    for g in checkpoints:
-        for key in UNNEST_KEYS:
-            if key not in g:
-                continue
-            unnest_dict = flatten_dict(g.pop(key))
-            g.update(unnest_dict)
-        g = flatten_dict(g)
-        checkpoint_dicts.append(g)
-
-    checkpoints_df = pd.DataFrame(checkpoint_dicts)
+    try:
+        checkpoints_df = ExperimentAnalysis(experiment_path).dataframe()
+    except TuneError:
+        print("No experiment state found!")
+        sys.exit(0)
 
     if not info_keys:
         info_keys = DEFAULT_EXPERIMENT_INFO_KEYS
@@ -241,19 +211,20 @@ def list_experiments(project_path,
     experiment_data_collection = []
 
     for experiment_dir in experiment_folders:
-        experiment_state = _get_experiment_state(
-            os.path.join(base, experiment_dir))
-        if not experiment_state:
+        analysis_obj, checkpoints_df = None, None
+        try:
+            analysis_obj = ExperimentAnalysis(
+                os.path.join(project_path, experiment_dir))
+            checkpoints_df = analysis_obj.dataframe()
+        except TuneError:
             logger.debug("No experiment state found in %s", experiment_dir)
             continue
 
-        checkpoints = pd.DataFrame(experiment_state["checkpoints"])
-        runner_data = experiment_state["runner_data"]
-
         # Format time-based values.
+        stats = analysis_obj.stats()
         time_values = {
-            "start_time": runner_data.get("_start_time"),
-            "last_updated": experiment_state.get("timestamp"),
+            "start_time": stats.get("_start_time"),
+            "last_updated": stats.get("timestamp"),
         }
 
         formatted_time_values = {
@@ -264,11 +235,12 @@ def list_experiments(project_path,
 
         experiment_data = {
             "name": experiment_dir,
-            "total_trials": checkpoints.shape[0],
-            "running_trials": (checkpoints["status"] == Trial.RUNNING).sum(),
+            "total_trials": checkpoints_df.shape[0],
+            "running_trials": (
+                checkpoints_df["status"] == Trial.RUNNING).sum(),
             "terminated_trials": (
-                checkpoints["status"] == Trial.TERMINATED).sum(),
-            "error_trials": (checkpoints["status"] == Trial.ERROR).sum(),
+                checkpoints_df["status"] == Trial.TERMINATED).sum(),
+            "error_trials": (checkpoints_df["status"] == Trial.ERROR).sum(),
         }
         experiment_data.update(formatted_time_values)
         experiment_data_collection.append(experiment_data)

python/ray/tune/ray_trial_executor.py

@@ -97,7 +97,8 @@ class RayTrialExecutor(TrialExecutor):
             # Set the working dir in the remote process, for user file writes
             if not os.path.exists(remote_logdir):
                 os.makedirs(remote_logdir)
-            os.chdir(remote_logdir)
+            if not ray.worker._mode() == ray.worker.LOCAL_MODE:
+                os.chdir(remote_logdir)
             return NoopLogger(config, remote_logdir)
 
         # Logging for trials is handled centrally by TrialRunner, so

python/ray/tune/tests/test_experiment_analysis.py

@@ -0,0 +1,125 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import unittest
+import shutil
+import tempfile
+import random
+import os
+import pandas as pd
+
+import ray
+from ray.tune import run, sample_from
+from ray.tune.analysis import ExperimentAnalysis
+from ray.tune.examples.async_hyperband_example import MyTrainableClass
+from ray.tune.schedulers import AsyncHyperBandScheduler
+
+
+class ExperimentAnalysisSuite(unittest.TestCase):
+    def setUp(self):
+        ray.init(local_mode=True)
+
+        self.test_dir = tempfile.mkdtemp()
+        self.test_name = "analysis_exp"
+        self.num_samples = 10
+        self.metric = "episode_reward_mean"
+        self.test_path = os.path.join(self.test_dir, self.test_name)
+        self.run_test_exp()
+
+        self.ea = ExperimentAnalysis(self.test_path)
+
+    def tearDown(self):
+        shutil.rmtree(self.test_dir, ignore_errors=True)
+        ray.shutdown()
+
+    def run_test_exp(self):
+        ahb = AsyncHyperBandScheduler(
+            time_attr="training_iteration",
+            reward_attr=self.metric,
+            grace_period=5,
+            max_t=100)
+
+        run(MyTrainableClass,
+            name=self.test_name,
+            scheduler=ahb,
+            local_dir=self.test_dir,
+            **{
+                "stop": {
+                    "training_iteration": 1
+                },
+                "num_samples": 10,
+                "config": {
+                    "width": sample_from(
+                        lambda spec: 10 + int(90 * random.random())),
+                    "height": sample_from(
+                        lambda spec: int(100 * random.random())),
+                },
+            })
+
+    def testDataframe(self):
+        df = self.ea.dataframe()
+
+        self.assertTrue(isinstance(df, pd.DataFrame))
+        self.assertEquals(df.shape[0], self.num_samples)
+
+    def testTrialDataframe(self):
+        cs = self.ea._checkpoints
+        idx = random.randint(0, len(cs) - 1)
+        trial_df = self.ea.trial_dataframe(
+            cs[idx]["trial_id"])  # random trial df
+
+        self.assertTrue(isinstance(trial_df, pd.DataFrame))
+        self.assertEqual(trial_df.shape[0], 1)
+
+    def testBestTrainable(self):
+        best_trainable = self.ea.get_best_trainable(self.metric,
+                                                    MyTrainableClass)
+
+        self.assertTrue(isinstance(best_trainable, MyTrainableClass))
+
+    def testBestConfig(self):
+        best_config = self.ea.get_best_config(self.metric)
+
+        self.assertTrue(isinstance(best_config, dict))
+        self.assertTrue("width" in best_config)
+        self.assertTrue("height" in best_config)
+
+    def testBestTrial(self):
+        best_trial = self.ea._get_best_trial(self.metric)
+
+        self.assertTrue(isinstance(best_trial, dict))
+        self.assertTrue("local_dir" in best_trial)
+        self.assertEqual(best_trial["local_dir"],
+                         os.path.expanduser(self.test_path))
+        self.assertTrue("config" in best_trial)
+        self.assertTrue("width" in best_trial["config"])
+        self.assertTrue("height" in best_trial["config"])
+        self.assertTrue("last_result" in best_trial)
+        self.assertTrue(self.metric in best_trial["last_result"])
+
+    def testCheckpoints(self):
+        checkpoints = self.ea._checkpoints
+
+        self.assertTrue(isinstance(checkpoints, list))
+        self.assertTrue(isinstance(checkpoints[0], dict))
+        self.assertEqual(len(checkpoints), self.num_samples)
+
+    def testStats(self):
+        stats = self.ea.stats()
+
+        self.assertTrue(isinstance(stats, dict))
+        self.assertTrue("start_time" in stats)
+        self.assertTrue("timestamp" in stats)
+
+    def testRunnerData(self):
+        runner_data = self.ea.runner_data()
+
+        self.assertTrue(isinstance(runner_data, dict))
+        self.assertTrue("_metadata_checkpoint_dir" in runner_data)
+        self.assertEqual(runner_data["_metadata_checkpoint_dir"],
+                         os.path.expanduser(self.test_path))
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

python/ray/tune/trial_runner.py

@@ -179,7 +179,10 @@ class TrialRunner(object):
             "checkpoints": list(
                 self.trial_executor.get_checkpoints().values()),
             "runner_data": self.__getstate__(),
-            "timestamp": time.time()
+            "stats": {
+                "start_time": self._start_time,
+                "timestamp": time.time()
+            }
         }
         tmp_file_name = os.path.join(metadata_checkpoint_dir,
                                      ".tmp_checkpoint")

python/ray/tune/visual_utils.py

@@ -14,6 +14,8 @@ from ray.tune.util import flatten_dict
 
 logger = logging.getLogger(__name__)
 
+logger.warning("This module will be deprecated in a future version of Tune.")
+
 
 def _parse_results(res_path):
     res_dict = {}