fix trainer/s3 bugs

Signed-off-by: Sunyanan Choochotkaew <[email protected]>

Author: sunya-ch

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/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)