Added support to retrieve HA config and prepared new version

.gitignore

@@ -1,5 +1,7 @@
 # ---> Python
 test.py
+# Legacy config yaml file
+config_emhass.yaml
 # Secret yaml file
 secrets_emhass.yaml
 # config json file 

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 0.11.2 - 2024-10-31
+### Improvement
+- Added support to retrieve HA configuration. This will be used in the future to automatically retrieve some parameters as the currency
+### Fix
+- utils fix runtime parameter merge bugs 
+- configuration_script.js fix placeholder value bug 
+
 ## 0.11.1 - 2024-10-29
 ### Fix
 - Fix parameter saving and duplicate battery bugs 

docs/conf.py

@@ -22,7 +22,7 @@
 author = 'David HERNANDEZ'
 
 # The full version, including alpha/beta/rc tags
-release = '0.11.1'
+release = '0.11.2'
 
 # -- General configuration ---------------------------------------------------
 

setup.py

@@ -19,7 +19,7 @@
 
 setup(
     name='emhass',  # Required
-    version='0.11.1',  # Required
+    version='0.11.2',  # Required
     description='An Energy Management System for Home Assistant',  # Optional
     long_description=long_description,  # Optional
     long_description_content_type='text/markdown',  # Optional (see note above)

src/emhass/retrieve_hass.py

@@ -72,6 +72,19 @@ def __init__(self, hass_url: str, long_lived_token: str, freq: pd.Timedelta,
         self.logger = logger
         self.get_data_from_file = get_data_from_file
 
+    def get_ha_config(self):
+        """
+        Extract some configuration data from HA.
+        
+        """
+        headers = {
+            "Authorization": "Bearer " + self.long_lived_token,
+            "content-type": "application/json"
+            }
+        url = self.hass_url+"api/config"
+        response_config = get(url, headers=headers)
+        self.ha_config = response_config.json()
+
     def get_data(self, days_list: pd.date_range, var_list: list, 
                  minimal_response: Optional[bool] = False, significant_changes_only: Optional[bool] = False,
                  test_url: Optional[str] = "empty") -> None:
@@ -98,15 +111,17 @@ def get_data(self, days_list: pd.date_range, var_list: list,
             are experimental
         """
         self.logger.info("Retrieve hass get data method initiated...")
+        headers = {
+            "Authorization": "Bearer " + self.long_lived_token,
+            "content-type": "application/json"
+            }
+        # Looping on each day from days list
         self.df_final = pd.DataFrame()
         x = 0  # iterate based on days
-        # Looping on each day from days list
         for day in days_list:
             for i, var in enumerate(var_list):
                 if test_url == "empty":
-                    if (
-                        self.hass_url == "http://supervisor/core/api"
-                    ):  # If we are using the supervisor API
+                    if (self.hass_url == "http://supervisor/core/api"):  # If we are using the supervisor API
                         url = (
                             self.hass_url
                             + "/history/period/"
@@ -124,16 +139,10 @@ def get_data(self, days_list: pd.date_range, var_list: list,
                         )
                     if minimal_response:  # A support for minimal response
                         url = url + "?minimal_response"
-                    if (
-                        significant_changes_only
-                    ):  # And for signicant changes only (check the HASS restful API for more info)
+                    if (significant_changes_only):  # And for signicant changes only (check the HASS restful API for more info)
                         url = url + "?significant_changes_only"
                 else:
                     url = test_url
-                headers = {
-                    "Authorization": "Bearer " + self.long_lived_token,
-                    "content-type": "application/json",
-                }
                 try:
                     response = get(url, headers=headers)
                 except Exception:

src/emhass/utils.py

@@ -644,9 +644,7 @@ def get_yaml_parse(params: str, logger: logging.Logger) -> Tuple[dict, dict, dic
         return False, False, False
 
     optim_conf = input_conf.get("optim_conf", {})
-
     retrieve_hass_conf = input_conf.get("retrieve_hass_conf", {})
-
     plant_conf = input_conf.get("plant_conf", {})
 
     # Format time parameters
@@ -659,6 +657,70 @@ def get_yaml_parse(params: str, logger: logging.Logger) -> Tuple[dict, dict, dic
 
     return retrieve_hass_conf, optim_conf, plant_conf
 
+def get_legacy_yaml_parse(emhass_conf: dict, use_secrets: Optional[bool] = True,
+                          params: Optional[str] = None) -> Tuple[dict, dict, dict]:
+    """
+    Perform parsing of the config.yaml file.
+    
+    :param emhass_conf: Dictionary containing the needed emhass paths
+    :type emhass_conf: dict
+    :param use_secrets: Indicate if we should use a secrets file or not.
+        Set to False for unit tests.
+    :type use_secrets: bool, optional
+    :param params: Configuration parameters passed from data/options.json
+    :type params: str
+    :return: A tuple with the dictionaries containing the parsed data
+    :rtype: tuple(dict)
+
+    """
+    if params is None:
+        with open(emhass_conf["config_path"], 'r') as file:
+            input_conf = yaml.load(file, Loader=yaml.FullLoader)
+    else:
+        input_conf = json.loads(params)
+    if use_secrets:
+        if params is None:
+            with open(emhass_conf["config_path"].parent / 'secrets_emhass.yaml', 'r') as file: # Assume secrets and config file paths are the same 
+                input_secrets = yaml.load(file, Loader=yaml.FullLoader)
+        else:
+            input_secrets = input_conf.pop("params_secrets", None)
+
+    if type(input_conf["retrieve_hass_conf"]) == list:  # if using old config version
+        retrieve_hass_conf = dict(
+            {key: d[key] for d in input_conf["retrieve_hass_conf"] for key in d}
+        )
+    else:
+        retrieve_hass_conf = input_conf.get("retrieve_hass_conf", {})
+
+    if use_secrets:
+        retrieve_hass_conf.update(input_secrets)
+    else:
+        retrieve_hass_conf["hass_url"] = "http://supervisor/core/api"
+        retrieve_hass_conf["long_lived_token"] = "${SUPERVISOR_TOKEN}"
+        retrieve_hass_conf["time_zone"] = "Europe/Paris"
+        retrieve_hass_conf["lat"] = 45.83
+        retrieve_hass_conf["lon"] = 6.86
+        retrieve_hass_conf["alt"] = 4807.8
+    retrieve_hass_conf["freq"] = pd.to_timedelta(retrieve_hass_conf["freq"], "minutes")
+    retrieve_hass_conf["time_zone"] = pytz.timezone(retrieve_hass_conf["time_zone"])
+
+    if type(input_conf["optim_conf"]) == list:
+        optim_conf = dict({key: d[key] for d in input_conf["optim_conf"] for key in d})
+    else:
+        optim_conf = input_conf.get("optim_conf", {})
+
+    optim_conf["list_hp_periods"] = dict(
+        (key, d[key]) for d in optim_conf["list_hp_periods"] for key in d
+    )
+    optim_conf["delta_forecast"] = pd.Timedelta(days=optim_conf["delta_forecast"])
+
+    if type(input_conf["plant_conf"]) == list:
+        plant_conf = dict({key: d[key] for d in input_conf["plant_conf"] for key in d})
+    else:
+        plant_conf = input_conf.get("plant_conf", {})
+
+    return retrieve_hass_conf, optim_conf, plant_conf
+
 
 def get_injection_dict(df: pd.DataFrame, plot_size: Optional[int] = 1366) -> dict:
     """
@@ -858,7 +920,7 @@ def build_config(emhass_conf: dict, logger: logging.Logger, defaults_path: str,
 
 
 def build_legacy_config_params(emhass_conf: dict, legacy_config: dict,
-                 logger: logging.Logger) -> dict:
+                               logger: logging.Logger) -> dict:
     """
     Build a config dictionary with legacy config_emhass.yaml file. 
     Uses the associations file to convert parameter naming conventions (to config.json/config_defaults.json).
@@ -909,8 +971,7 @@ def build_legacy_config_params(emhass_conf: dict, legacy_config: dict,
     return config
     # params['associations_dict'] = associations_dict
 
-def param_to_config(param: dict,
-                 logger: logging.Logger) -> dict:
+def param_to_config(param: dict, logger: logging.Logger) -> dict:
     """
     A function that extracts the parameters from param back to the config.json format.
     Extracts parameters from config catagories.
@@ -940,7 +1001,7 @@ def param_to_config(param: dict,
     return return_config
 
 def build_secrets(emhass_conf: dict, logger: logging.Logger, argument: Optional[dict] = {}, options_path: Optional[str] = None, 
-                 secrets_path: Optional[str] = None, no_response: Optional[bool] = False) -> Tuple[dict, dict]:    
+                  secrets_path: Optional[str] = None, no_response: Optional[bool] = False) -> Tuple[dict, dict]:    
     """
     Retrieve and build parameters from secrets locations (ENV, ARG, Secrets file (secrets_emhass.yaml/options.json) and/or Home Assistant (via API))
     priority order (lwo to high) = Defaults (written in function), ENV, Options json file, Home Assistant API,  Secrets yaml file, Arguments

tests/test_retrieve_hass.py

@@ -34,10 +34,16 @@ def setUp(self):
         save_data_to_file = False
 
         # Build params with default secrets (no config)
-        if emhass_conf['defaults_path'].exists():   
-            _,secrets = utils.build_secrets(emhass_conf,logger,no_response=True)
-            params =  utils.build_params(emhass_conf,secrets,{},logger)
-            retrieve_hass_conf, _, _ = get_yaml_parse(params,logger)
+        if emhass_conf['defaults_path'].exists():
+            if get_data_from_file:
+                _,secrets = utils.build_secrets(emhass_conf,logger,no_response=True)
+                params =  utils.build_params(emhass_conf,secrets,{},logger)
+                retrieve_hass_conf, _, _ = get_yaml_parse(params,logger)
+            else:
+                emhass_conf['secrets_path'] = root / 'secrets_emhass.yaml'
+                emhass_conf['config_path'] = pathlib.Path(root) / 'config_emhass.yaml'
+                retrieve_hass_conf, _, _ = utils.get_legacy_yaml_parse(emhass_conf)
+                params = None
         else:
             raise Exception("config_defaults. does not exist in path: "+str(emhass_conf['defaults_path'] ))
 
@@ -60,8 +66,8 @@ def setUp(self):
                 self.rh.df_final, self.days_list, self.var_list = pickle.load(inp)
         # Else obtain sensor values from HA
         else:
-            self.days_list = get_days_list(self.retrieve_hass_conf['historic_days_to_retrieve'])
-            self.var_list = [self.retrieve_hass_conf['sensor_power_load_no_var_loads'], self.retrieve_hass_conf['sensor_power_photovoltaics']]
+            self.days_list = get_days_list(self.retrieve_hass_conf['days_to_retrieve'])
+            self.var_list = [self.retrieve_hass_conf['var_PV'], self.retrieve_hass_conf['var_load']]
             self.rh.get_data(self.days_list, self.var_list,
                              minimal_response=False, significant_changes_only=False)
             # Check to save updated data to file