Merge pull request #229 from sunya-ch/fix-bugs

Add metadata plot and Fix bugs

Author: rootfs

cmd/cmd_plot.py

@@ -111,3 +111,39 @@ def summary_plot(args, energy_source, summary_df, output_folder, name):
     filename = os.path.join(output_folder, name + ".png")
     fig.savefig(filename)
     plt.close()
+
+def metadata_plot(args, energy_source, metadata_df, output_folder, name):
+    if metadata_df is None or len(metadata_df) == 0:
+        print("no metadata data to plot")
+        return
+
+    plot_height = 5
+    plot_width = 20
+
+    import matplotlib.pyplot as plt
+    import seaborn as sns
+    sns.set(font_scale=1.2)
+
+    energy_components = PowerSourceMap[energy_source]
+    col_num = len(energy_components)
+    fig, axes = plt.subplots(col_num, 1, figsize=(plot_width, plot_height*col_num))
+    for i in range(0, col_num):
+        component = energy_components[i]
+        metadata_df = metadata_df.sort_values(by="feature_group")
+        if col_num == 1:
+            ax = axes
+        else:
+            ax = axes[i]
+        sns.boxplot(data=metadata_df, x="feature_group", y="mae", hue="trainer", ax=ax, hue_order=sorted(metadata_df['trainer'].unique()), showfliers=False, palette="Set3")
+        ax.set_title(component)
+        ax.set_ylabel("MAE (Watt)")
+        ax.set_xlabel("Feature Group")
+        # ax.set_ylim((0, 100))
+        if i < col_num-1:
+            ax.set_xlabel("")
+       #  ax.legend(bbox_to_anchor=(1.05, 1.05))
+    plt.suptitle("Pipieline metadata of {} {}".format(energy_source.upper(), args.output_type))
+    plt.tight_layout()
+    filename = os.path.join(output_folder, name + ".png")
+    fig.savefig(filename)
+    plt.close()

cmd/cmd_util.py

@@ -240,16 +240,20 @@ def check_ot_fg(args, valid_fg):
             exit()
     return ot, fg
 
+import sklearn
 def assert_train(trainer, data, energy_components):
     import pandas as pd
     node_types = pd.unique(data[node_info_column])
     for node_type in node_types:
         node_type_filtered_data = data[data[node_info_column] == node_type]
         X_values = node_type_filtered_data[trainer.features].values
         for component in energy_components:
-            output = trainer.predict(node_type, component, X_values)
-            if output is not None:
-                assert len(output) == len(X_values), "length of predicted values != features ({}!={})".format(len(output), len(X_values))
+            try:
+                output = trainer.predict(node_type, component, X_values)
+                if output is not None:
+                    assert len(output) == len(X_values), "length of predicted values != features ({}!={})".format(len(output), len(X_values))
+            except sklearn.exceptions.NotFittedError:
+                pass
 
 def get_isolator(data_path, isolator, profile, pipeline_name, target_hints, bg_hints, abs_pipeline_name, replace_node_type=default_node_type):
     pipeline_path = get_pipeline_path(data_path, pipeline_name=pipeline_name)

cmd/main.py

@@ -19,15 +19,15 @@
 from util.extract_types import get_expected_power_columns
 from util.train_types import ModelOutputType, FeatureGroups, is_single_source_feature_group, all_feature_groups, default_trainers
 from util.loader import load_json, DEFAULT_PIPELINE, load_pipeline_metadata, get_pipeline_path, get_model_group_path, list_pipelines, list_model_names, load_metadata, load_csv, get_preprocess_folder, get_general_filename, load_machine_spec
-from util.saver import save_json, save_csv, save_train_args
+from util.saver import save_json, save_csv, save_train_args, _pipeline_model_metadata_filename
 from util.config import ERROR_KEY, model_toppath
 from util import get_valid_feature_group_from_queries, PowerSourceMap
 from train.prom.prom_query import _range_queries
 from train.exporter import exporter
 from train import load_class
 from train.profiler.node_type_index import NodeTypeIndexCollection, NodeTypeSpec, generate_spec
 
-from cmd_plot import ts_plot, feature_power_plot, summary_plot
+from cmd_plot import ts_plot, feature_power_plot, summary_plot, metadata_plot
 from cmd_util import extract_time, save_query_results, get_validate_df, summary_validation, get_extractor, check_ot_fg, get_pipeline, assert_train, get_isolator, UTC_OFFSET_TIMEDELTA
 
 import threading
@@ -597,9 +597,11 @@ def estimate(args):
     - `preprocess` plots time series of usage and power metrics for both AbsPower and DynPower
     - `estimate` passes all arguments to `estimate` function, and plots the predicted time series and correlation between usage and power metrics
     - `error` passes all arguments to `estimate` function, and plots the summary of prediction error
+    - `metadata` plot pipeline metadata 
 - --energy-source : specify target energy sources (use comma(,) as delimiter) 
 - --extractor : specify extractor to get preprocessed data of AbsPower model linked to the input data
 - --isolator : specify isolator to get preprocessed data of DynPower model linked to the input data
+- --pipeline_name : specify pipeline name
 """
 
 def plot(args):
@@ -685,6 +687,12 @@ def plot(args):
         for energy_source in energy_sources:
             data_filename = get_general_filename(args.target_data, energy_source, fg, ot, args.extractor, args.isolator)
             summary_plot(args, energy_source, summary_df, output_folder, data_filename)
+    elif args.target_data == "metadata":
+        for energy_source in energy_sources:
+            data_filename = _pipeline_model_metadata_filename(energy_source, ot.name)
+            pipeline_path = get_pipeline_path(data_path, pipeline_name=pipeline_name)
+            model_metadata_df = load_pipeline_metadata(pipeline_path, energy_source, ot.name)
+            metadata_plot(args, energy_source, model_metadata_df, output_folder, data_filename)
 
 """
 export
@@ -732,7 +740,15 @@ def export(args):
     pipeline_name = args.pipeline_name
     pipeline_path = get_pipeline_path(data_path, pipeline_name=pipeline_name)
 
-    exporter.export(data_path, pipeline_path, output_path, publisher=args.publisher, collect_date=collect_date, inputs=inputs)
+    local_export_path = exporter.export(data_path, pipeline_path, output_path, publisher=args.publisher, collect_date=collect_date, inputs=inputs)
+    args.target_data = "metadata"
+
+    args.output = local_export_path
+    args.output_type = "AbsPower"
+    args.energy_source = ",".join(PowerSourceMap.keys())
+    plot(args)
+    args.output_type = "DynPower"
+    plot(args)
 
 """
 plot_scenario

model_training/s3/s3-loader.py

@@ -20,6 +20,7 @@ def ibmcloud_list_keys(client, bucket_name, prefix):
 
 def get_bucket_file_map(client, bucket_name, machine_id, mnt_path, pipeline_name, list_func):
     bucket_file_map = dict()
+    top_key_path = ""
     if machine_id is not None and machine_id != "":
         top_key_path = "/" + machine_id
     # add data key map

src/train/exporter/exporter.py

@@ -11,6 +11,7 @@
 
 from validator import get_validated_export_items, BestModelCollection
 from loader import load_metadata, load_node_type_index, get_version_path, get_export_path
+from saver import save_pipeline_metadata, save_node_type_index
 from format import time_to_str
 from writer import generate_pipeline_page, generate_report_results, generate_pipeline_readme, append_version_readme, get_workload_content
 from config import ERROR_KEY
@@ -44,7 +45,14 @@ def export(data_path, pipeline_path, db_path, publisher, collect_date, inputs):
     remote_version_path = get_version_path(repo_url, assure=False)
 
     # get validated export items (models)
-    export_items = get_validated_export_items(pipeline_path, pipeline_name)
+    export_items, valid_metadata_df = get_validated_export_items(pipeline_path, pipeline_name)
+    # save pipeline metadata
+    for energy_source, ot_metadata_df in valid_metadata_df.items():
+        for model_type, metadata_df in ot_metadata_df.items():
+            metadata_df = metadata_df.sort_values(by=["feature_group", ERROR_KEY])
+            save_pipeline_metadata(local_export_path, pipeline_metadata, energy_source, model_type, metadata_df)
+    # save node_type_index.json
+    save_node_type_index(local_export_path, node_type_index_json)
 
     for export_item in export_items:
         # export models
@@ -60,4 +68,6 @@ def export(data_path, pipeline_path, db_path, publisher, collect_date, inputs):
     # generate validation result page
     generate_pipeline_readme(pipeline_name, local_export_path, node_type_index_json, best_model_collections)
     # add new pipeline item to version path
-    append_version_readme(local_version_path, pipeline_metadata)
+    append_version_readme(local_version_path, pipeline_metadata)
+
+    return local_export_path

src/train/exporter/validator.py

@@ -88,13 +88,16 @@ def get_best_item_with_weight(self, energy_source, output_type_name, feature_gro
 # get_validated_export_items return valid export items
 def get_validated_export_items(pipeline_path, pipeline_name):
     export_items = []
+    valid_metadata_df = dict()
     models_path = os.path.join(pipeline_path, "..")
     for energy_source in PowerSourceMap.keys():
+        valid_metadata_df[energy_source] = dict()
         for ot in ModelOutputType:
             metadata_df = load_pipeline_metadata(pipeline_path, energy_source, ot.name)
             if metadata_df is None:
                 print("no metadata for", energy_source, ot.name)
                 continue
+            valid_rows = []
             for _, row in metadata_df.iterrows():
                 if row['mape'] <= mape_threshold or row['mae'] <= mae_threshold:
                     model_name = row["model_name"]
@@ -105,4 +108,6 @@ def get_validated_export_items(pipeline_path, pipeline_name):
                         print("source not exist: ", source_file)
                         continue
                     export_items += [export_item]
-    return export_items
+                    valid_rows += [row]
+            valid_metadata_df[energy_source][ot.name] = pd.DataFrame(valid_rows)
+    return export_items, valid_metadata_df

src/train/exporter/writer.py

@@ -7,8 +7,9 @@
 sys.path.append(util_path)
 
 from loader import load_json, version
-from saver import assure_path
+from saver import assure_path,  _pipeline_model_metadata_filename
 from validator import mae_threshold, mape_threshold
+from train_types import ModelOutputType, PowerSourceMap
 
 error_report_foldername = "error_report"
 
@@ -201,7 +202,6 @@ def generate_pipeline_readme(pipeline_name, local_export_path, node_type_index_j
     markdown_filepath = os.path.join(local_export_path, "README.md")
     markdown_content = "# {} on v{} Build\n\n".format(pipeline_name, version)
     markdown_content += "MAE Threshold = {}, MAPE Threshold = {}%\n\n".format(mae_threshold, int(mape_threshold))
-    
     items = []
     for node_type, spec_json in node_type_index_json.items():
         if best_model_collections[int(node_type)].has_model:
@@ -213,6 +213,12 @@ def generate_pipeline_readme(pipeline_name, local_export_path, node_type_index_j
             items += [item]
     df = pd.DataFrame(items)
     markdown_content += "Available Node Type: {}\n\n".format(len(df))
+    # add metadata figures
+    for ot in ModelOutputType:
+        for energy_source in PowerSourceMap.keys():
+            data_filename = _pipeline_model_metadata_filename(energy_source, ot.name)
+            markdown_content += "![]({}.png)\n".format(data_filename)
+
     markdown_content += data_to_markdown_table(df.sort_values(by=["node type"]))
     write_markdown(markdown_filepath, markdown_content)
     return markdown_filepath

src/train/pipeline.py

@@ -117,7 +117,7 @@ def prepare_data_from_input_list(self, input_query_results_list, energy_componen
 
     def _train(self, abs_data, dyn_data, power_labels, energy_source, feature_group):
         # start the thread pool
-        with ThreadPoolExecutor(2) as executor:
+        with ThreadPoolExecutor(len(self.trainers)) as executor:
             futures = []
             for trainer in self.trainers:
                 if trainer.feature_group_name != feature_group:

src/train/profiler/node_type_index.py

@@ -44,7 +44,7 @@ def generate_spec(data_path, machine_id):
     if "brand_raw" in cpu_info:
         processor = format_processor(cpu_info["brand_raw"])
     cores = psutil.cpu_count(logical=True)
-    chips = psutil.cpu_count(logical=False)
+    chips = int(cores/psutil.cpu_count(logical=False))
     memory = psutil.virtual_memory().total
     memory_gb = int(memory/GB)
     cpu_freq_mhz = round(psutil.cpu_freq(percpu=False).max/100)*100 # round to one decimal of GHz

src/train/trainer/__init__.py

@@ -130,39 +130,41 @@ def load_model(self, node_type):
 
     def process(self, data, power_labels, pipeline_lock):
         node_types = pd.unique(data[node_info_column])
-        try:
-            for node_type in node_types:
-                node_type = int(node_type)
-                save_path = self._get_save_path(str(node_type))
-                self.node_scalers[node_type] = load_scaler(save_path)
-                self.load_model(node_type)
-                
-                node_type_filtered_data = data[data[node_info_column] == node_type]
-                if self.node_scalers[node_type] is None:
-                    self.print_log("fit scaler to latest data {1} for node_type={0}".format(node_type, self.feature_group_name))
-                    # no profiled scaler
-                    x_values = node_type_filtered_data[self.features].values
-                    self.node_scalers[node_type] = MaxAbsScaler()
-                    self.node_scalers[node_type].fit(x_values)
-                
-                X_test_map = dict()
-                y_test_map = dict()
+        for node_type in node_types:
+            node_type = int(node_type)
+            save_path = self._get_save_path(str(node_type))
+            self.node_scalers[node_type] = load_scaler(save_path)
+            self.load_model(node_type)
+            
+            node_type_filtered_data = data[data[node_info_column] == node_type]
+            if self.node_scalers[node_type] is None:
+                self.print_log("fit scaler to latest data {1} for node_type={0}".format(node_type, self.feature_group_name))
+                # no profiled scaler
+                x_values = node_type_filtered_data[self.features].values
+                self.node_scalers[node_type] = MaxAbsScaler()
+                self.node_scalers[node_type].fit(x_values)
+            
+            X_test_map = dict()
+            y_test_map = dict()
+            try:
                 for component in self.energy_components:
                     X_values, y_values = self.apply_ratio(component, node_type_filtered_data, power_labels)
                     X_train, X_test, y_train, y_test = normalize_and_split(X_values, y_values, scaler=self.node_scalers[node_type])
                     X_test_map[component] = X_test
                     y_test_map[component] = y_test
                     self.train(node_type, component, X_train, y_train)
                     self.save_checkpoint(self.node_models[node_type][component], self._checkpoint_filepath(component, node_type))
-                if self.should_archive(node_type):
-                    pipeline_lock.acquire()
-                    try:
-                        self.save_model_and_metadata(node_type, X_test_map, y_test_map)
-                    finally:
-                        pipeline_lock.release()
-        except Exception as e:
-            print(e)
-            pipeline_lock.release()
+            except Exception as err:
+                self.print_log("failed to process {}: {}".format(node_type, err))
+                continue
+            if self.should_archive(node_type):
+                pipeline_lock.acquire()
+                try:
+                    self.save_model_and_metadata(node_type, X_test_map, y_test_map)
+                except Exception as err:
+                    self.print_log("failed to save model {}: {}".format(node_type, err))
+                finally:
+                    pipeline_lock.release()
 
     def apply_ratio(self, component, node_type_filtered_data, power_labels):
         power_label = component_to_col(component)