feature: export experiment results

docs/en_US/Tutorial/Nnictl.md

@@ -24,6 +24,7 @@ nnictl support commands:
 * [nnictl package](#package)
 * [nnictl ss_gen](#ss_gen)
 * [nnictl --version](#version)
+* [nnictl export_results](#export-results)
 
 ### Manage an experiment
 
@@ -465,13 +466,14 @@ Debug mode will disable version check function in Trialkeeper.
   |id|  False| |ID of the experiment    |
   |--filename, -f|  True| |File path of the output file     |
   |--type|  True| |Type of output file, only support "csv" and "json"|
+  |--intermediate, -i|False||Is intermediate results required|
 
   * Examples
 
   > export all trial data in an experiment as json format
 
   ```bash
-  nnictl experiment export [experiment_id] --filename [file_path] --type json
+  nnictl experiment export [experiment_id] --filename [file_path] --type json --intermediate True
   ```
 
 * __nnictl experiment import__
@@ -850,4 +852,3 @@ Debug mode will disable version check function in Trialkeeper.
     ```bash
     nnictl --version
     ```
-

tools/nni_cmd/nnictl.py

@@ -148,6 +148,7 @@ def parse_args():
     parser_trial_export.add_argument('id', nargs='?', help='the id of experiment')
     parser_trial_export.add_argument('--type', '-t', choices=['json', 'csv'], required=True, dest='type', help='target file type')
     parser_trial_export.add_argument('--filename', '-f', required=True, dest='path', help='target file path')
+    parser_trial_export.add_argument('--intermediate', '-i', default=False, help='is intermediate results required')
     parser_trial_export.set_defaults(func=export_trials_data)
 
     #TODO:finish webui function

tools/nni_cmd/nnictl_utils.py

@@ -9,14 +9,15 @@
 import re
 import shutil
 import subprocess
+from functools import reduce
 from datetime import datetime, timezone
 from pathlib import Path
 from subprocess import Popen
 from pyhdfs import HdfsClient
 from nni.package_utils import get_nni_installation_path
 from nni_annotation import expand_annotations
 from .rest_utils import rest_get, rest_delete, check_rest_server_quick, check_response
-from .url_utils import trial_jobs_url, experiment_url, trial_job_id_url, export_data_url
+from .url_utils import trial_jobs_url, experiment_url, trial_job_id_url, export_data_url, metric_data_url
 from .config_utils import Config, Experiments
 from .constants import NNICTL_HOME_DIR, EXPERIMENT_INFORMATION_FORMAT, EXPERIMENT_DETAIL_FORMAT, \
      EXPERIMENT_MONITOR_INFO, TRIAL_MONITOR_HEAD, TRIAL_MONITOR_CONTENT, TRIAL_MONITOR_TAIL, REST_TIME_OUT
@@ -681,30 +682,53 @@ def monitor_experiment(args):
     set_monitor(False, args.time)
 
 def export_trials_data(args):
-    '''export experiment metadata to csv
+    '''export experiment metadata and intermediate results to json or csv
     '''
+    def groupby_trial_id(intermediate_results):
+        sorted(intermediate_results, key=lambda x: x['timestamp'])
+        groupby = dict()
+        for content in intermediate_results:
+            groupby.setdefault(content['trialJobId'], []).append(content['data'][2:-2])
+        return groupby
+
+    def trans_intermediate_dict(record):
+        return {'intermediate': str(reduce(lambda x, y: x + y, record))}
+
     nni_config = Config(get_config_filename(args))
     rest_port = nni_config.get_config('restServerPort')
     rest_pid = nni_config.get_config('restServerPid')
+    print(vars(args))
     if not detect_process(rest_pid):
         print_error('Experiment is not running...')
         return
     running, response = check_rest_server_quick(rest_port)
     if running:
         response = rest_get(export_data_url(rest_port), 20)
+        if args.intermediate:
+            intermediate_results = rest_get(metric_data_url(rest_port), REST_TIME_OUT)
+            if not intermediate_results or not check_response(intermediate_results):
+                print_error('Error getting intermediate results.')
+                return
+            intermediate_results = groupby_trial_id(json.loads(intermediate_results.text))
         if response is not None and check_response(response):
+            content = json.loads(response.text)
+            if args.intermediate:
+                    for record in content:
+                        record['intermediate'] = intermediate_results[record['id']]
             if args.type == 'json':
                 with open(args.path, 'w') as file:
-                    file.write(response.text)
+                    file.write(json.dumps(content))
             elif args.type == 'csv':
-                content = json.loads(response.text)
                 trial_records = []
                 for record in content:
+                    print(record)
                     record_value = json.loads(record['value'])
                     if not isinstance(record_value, (float, int)):
-                        formated_record = {**record['parameter'], **record_value, **{'id': record['id']}}
+                        formated_record = {**record['parameter'], **record_value, **{'id': record['id']},
+                                           **trans_intermediate_dict(record['intermediate'])}
                     else:
-                        formated_record = {**record['parameter'], **{'reward': record_value, 'id': record['id']}}
+                        formated_record = {**record['parameter'], **{'reward': record_value, 'id': record['id']},
+                                           **trans_intermediate_dict(record['intermediate'])}
                     trial_records.append(formated_record)
                 if not trial_records:
                     print_error('No trial results collected! Please check your trial log...')
@@ -736,3 +760,64 @@ def search_space_auto_gen(args):
         print_warning('Expected search space file \'{}\' generated, but not found.'.format(file_path))
     else:
         print_normal('Generate search space done: \'{}\'.'.format(file_path))
+
+def export_results(args):
+    '''dump all intermediate results and final results to json file
+    '''
+
+    def groupby_trial_id(intermediate_results):
+        sorted(intermediate_results, key=lambda x: x['timestamp'])
+        groupby = dict()
+        for content in intermediate_results:
+            groupby.setdefault(content['trialJobId'], []).append(content)
+        return groupby
+
+    update_experiment()
+
+    nni_config = Config(get_config_filename(args))
+    rest_port = nni_config.get_config('restServerPort')
+    rest_pid = nni_config.get_config('restServerPid')
+    if not detect_process(rest_pid):
+        print_error('Experiment is not running...')
+        return
+    running, _ = check_rest_server_quick(rest_port)
+    if running:
+        experiment_info = rest_get(experiment_url(rest_port), REST_TIME_OUT)
+        trial_info = rest_get(metric_data_url(rest_port), REST_TIME_OUT)
+        trial_jobs = rest_get(trial_jobs_url(rest_port), REST_TIME_OUT)
+
+        if experiment_info and check_response(experiment_info) and \
+                trial_info and check_response(trial_info) and \
+                trial_jobs and check_response(trial_jobs):
+            results = {}
+            experiment_info = json.loads(experiment_info.text)
+            results['experimentParameters'] = experiment_info
+            experiment_id = experiment_info.get('id')
+            trial_info = json.loads(trial_info.text)
+            group_trial_info = groupby_trial_id(trial_info)
+            trial_jobs = json.loads(trial_jobs.text)
+            trial_message = []
+            for trial in trial_jobs:
+                trial_id = trial['id']
+                trial['intermediate'] = group_trial_info[trial_id]
+                trial_message.append(trial)
+            results['trialMessage'] = trial_message
+        else:
+            print_error('Ops. Error occured connecting to the server.')
+            exit(1)
+    else:
+        args = vars(args)
+        experiment_config = Experiments()
+        experiment_dict = experiment_config.get_all_experiments()
+        print_error('Ops. The experiment to export isn\'t running now. Please use \n    \
+        nnictl resume {}\nto restart.'.format(args.id if args.id in experiment_dict else ''))
+        exit(1)
+    args = vars(args)
+    filename = 'exp_' + str(experiment_id) + '_' + str(time.time()) + '.json' \
+        if args.get('name') == '' else args.get('name')
+    if os.path.exists(filename):
+        print_error('File {0} has existed.'.format(filename))
+        exit(1)
+    with open(filename, 'w') as f:
+        f.write(json.dumps(results))
+    print_normal('Export expriment {0} done: {1}'.format(experiment_id, filename))

tools/nni_cmd/url_utils.py

@@ -22,6 +22,12 @@
 
 TENSORBOARD_API = '/tensorboard'
 
+METRIC_DATA_API = '/metric-data'
+
+def metric_data_url(port):
+    '''get metric_data url'''
+    return '{0}:{1}{2}{3}'.format(BASE_URL, port, API_ROOT_URL, METRIC_DATA_API)
+
 
 def check_status_url(port):
     '''get check_status url'''