fix(config): use predictable paths and proper typings

This modifies config to make use of proper python typing for config and
to make use of predictable paths for directories

Changes include:
* adding new RESOURCE_DIR config for resources dir
* add typing info to get_config to return proper types
* fix casing to follow pep8

Signed-off-by: Sunil Thaha <[email protected]>

Author: Sunil Thaha

src/kepler_model/estimate/estimator.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import os
+import pathlib
 import shutil
 import signal
 import socket
@@ -188,11 +189,11 @@ def sig_handler(signum, frame) -> None:
 @click.option(
     "--config-dir",
     "-c",
-    type=click.Path(exists=False, dir_okay=True, file_okay=False),
+    type=click.Path(exists=False, dir_okay=True, file_okay=False, path_type=pathlib.Path),
     default=CONFIG_PATH,
     required=False,
 )
-def run(log_level: str, machine_spec: str, config_dir: str) -> int:
+def run(log_level: str, machine_spec: str, config_dir: pathlib.Path) -> int:
     level = getattr(logging, log_level.upper())
     logging.basicConfig(
         level=level,

src/kepler_model/server/model_server.py

@@ -2,6 +2,7 @@
 import enum
 import logging
 import os
+import pathlib
 import shutil
 import sys
 
@@ -16,7 +17,7 @@
     MODEL_SERVER_MODEL_LIST_PATH,
     MODEL_SERVER_MODEL_REQ_PATH,
     download_path,
-    getConfig,
+    get_config,
     initial_pipeline_urls,
     model_toppath,
     set_config_dir,
@@ -102,7 +103,7 @@ class ModelListParam(enum.Enum):
 
 
 ###########################################
-MODEL_SERVER_PORT = int(getConfig("MODEL_SERVER_PORT", "8100"))
+MODEL_SERVER_PORT = get_config("MODEL_SERVER_PORT", 8100)
 
 # pipelineName and nodeCollection are global dict values set at initial state (load_init_pipeline)
 ## pipelineName: map of energy_source to target pipeline name
@@ -190,9 +191,9 @@ def select_best_model(
                 continue
         else:
             response = get_archived_file(valid_group_path, model_name)
-            if not os.path.exists(response):
+            if response.exists() is False:
                 # archived model file does not exists
-                logger.warning(f"archive failed: {response}")
+                logger.warning(f"archive file for {model_name}: {response} does not exist")
                 continue
         if best_cadidate is None or best_cadidate[ERROR_KEY] > metadata[ERROR_KEY]:
             best_cadidate = metadata
@@ -341,8 +342,7 @@ def unpack_zip_files(root_folder):
                 zip_file_path = os.path.join(folder, file)
                 extract_to = os.path.splitext(zip_file_path)[0]  # Extract to same location as the ZIP file
                 # Make sure the destination folder exists, if not, create it
-                if not os.path.exists(extract_to):
-                    os.makedirs(extract_to)
+                os.makedirs(extract_to, exist_ok=True)
                 shutil.unpack_archive(zip_file_path, extract_to)
                 weight_file = os.path.join(folder, file.replace(".zip", ".json"))
                 if os.path.exists(weight_file):
@@ -435,11 +435,11 @@ def fill_machine_spec():
 @click.option(
     "--config-dir",
     "-c",
-    type=click.Path(exists=False, dir_okay=True, file_okay=False),
+    type=click.Path(exists=False, dir_okay=True, file_okay=False, path_type=pathlib.Path),
     default=CONFIG_PATH,
     required=False,
 )
-def run(log_level: str, config_dir: str) -> int:
+def run(log_level: str, config_dir: pathlib.Path) -> int:
     level = getattr(logging, log_level.upper())
     logging.basicConfig(
         level=level,

src/kepler_model/train/exporter/validator.py

@@ -22,7 +22,7 @@ def __init__(self, models_path, output_type, feature_group, energy_source, pipel
         self.feature_group = feature_group
         self.model_name = model_name
         self.source_model_group_path = get_model_group_path(models_path, output_type, feature_group, energy_source, assure=False, pipeline_name=pipeline_name)
-        self.source_model_path = os.path.join(self.source_model_group_path, self.model_name)
+        self.source_model_path = self.source_model_group_path / self.model_name
         self.source_model_zip = get_archived_file(self.source_model_group_path, self.model_name)
         self.metadata = metadata
         self.node_type = metadata["node_type"]

src/kepler_model/train/online_trainer.py

@@ -6,14 +6,12 @@
 from kepler_model.train.pipeline import NewPipeline
 from kepler_model.train.profiler.profiler import load_all_profiles
 from kepler_model.train.prom.prom_query import PrometheusClient
-from kepler_model.util.config import getConfig
+from kepler_model.util.config import get_config
 from kepler_model.util.loader import default_train_output_pipeline
 from kepler_model.util.prom_types import PROM_QUERY_INTERVAL, get_valid_feature_group_from_queries
 from kepler_model.util.train_types import FeatureGroups, PowerSourceMap
 
-SAMPLING_INTERVAL = PROM_QUERY_INTERVAL
-SAMPLING_INTERVAL = getConfig("SAMPLING_INTERVAL", SAMPLING_INTERVAL)
-SAMPLING_INTERVAL = int(SAMPLING_INTERVAL)
+SAMPLING_INTERVAL = get_config("SAMPLING_INTERVAL", PROM_QUERY_INTERVAL)
 
 
 default_trainers = ["GradientBoostingRegressorTrainer"]

src/kepler_model/train/pipeline.py

@@ -209,11 +209,11 @@ def print_pipeline_process_end(self, energy_source, feature_group, abs_data, dyn
         print_bounded_multiline_message(messages)
 
     def archive_pipeline(self):
-        save_path = os.path.join(model_toppath, self.name)
+        save_path = model_toppath / self.name
         archived_file = get_archived_file(model_toppath, self.name)
-        self.print_log("archive pipeline :" + archived_file)
-        self.print_log("save_path :" + save_path)
-        shutil.make_archive(save_path, "zip", save_path)
+        self.print_log(f"archive pipeline : {archived_file}")
+        self.print_log(f"save_path : {save_path}")
+        shutil.make_archive(str(save_path), "zip", save_path)
 
 
 def initial_trainers(trainer_names, node_level, pipeline_name, target_energy_sources, valid_feature_groups):

src/kepler_model/train/profiler/profiler.py

@@ -15,11 +15,13 @@
 
 import json
 import os
+import pathlib
 from urllib.request import urlopen
 
 import joblib
 import pandas as pd
 
+from kepler_model.util import config
 from kepler_model.util.extract_types import component_to_col
 from kepler_model.util.loader import default_node_type
 from kepler_model.util.prom_types import generate_dataframe_from_response, node_info_column, node_info_query
@@ -30,23 +32,13 @@
 max_watt_key = "max_watt"
 unit_num_key = "#unit"
 
-resource_path = os.path.join(os.path.dirname(__file__), "..", "..", "..", "resource")
-default_profile_top_path = os.path.join(resource_path, "profiles")
-
-if not os.path.exists(resource_path):
-    os.mkdir(resource_path)
-
+default_profile_top_path = config.RESOURCE_DIR / "profiles"
 profiler_registry = "https://raw.githubusercontent.com/sustainable-computing-io/kepler-model-db/main/profiles"
 
 
-def prepare_profile_path(profile_top_path):
-    if not os.path.exists(profile_top_path):
-        os.mkdir(profile_top_path)
-
-    profile_path = os.path.join(profile_top_path, "profile")
-    if not os.path.exists(profile_path):
-        os.mkdir(profile_path)
-
+def prepare_profile_path(profile_top_path: str | pathlib.Path) -> pathlib.Path:
+    profile_path = pathlib.Path(profile_top_path) / "profile"
+    os.makedirs(profile_path, exist_ok=True)
     return profile_path
 
 

src/kepler_model/train/trainer/__init__.py

@@ -238,9 +238,9 @@ def archive_model(self, node_type):
         save_path = self._get_save_path(node_type)
         model_name, _ = self._model_filename(node_type)
         archived_file = get_archived_file(self.group_path, model_name)
-        self.print_log("archive model :" + archived_file)
-        self.print_log("save_path :" + save_path)
-        shutil.make_archive(save_path, "zip", save_path)
+        self.print_log(f"archive model :  {archived_file}")
+        self.print_log(f"save_path : {save_path}")
+        shutil.make_archive(str(save_path), "zip", save_path)
         weight_dict = self.get_weight_dict(node_type)
         if weight_dict is not None:
             save_weight(save_path, weight_dict)

src/kepler_model/util/__init__.py

@@ -1,5 +1,5 @@
 # commonly-used definitions
-from .config import getConfig, model_toppath
+from .config import get_config, model_toppath
 from .loader import (
     class_to_json,
     default_train_output_pipeline,
@@ -47,7 +47,7 @@
     "save_metadata",
     "save_scaler",
     "save_weight",
-    "getConfig",
+    "get_config",
     "model_toppath",
     "SYSTEM_FEATURES",
     "COUNTER_FEAUTRES",