Feat - Improved coverage and added support for optimization status pu…

…blish

CHANGELOG.md

@@ -3,8 +3,10 @@
 ## [0.6.0] - 2023-12-16
 ### Improvement
 - Now Python 3.11 is fully supported, thanks to @pail23
+- We now publish the optimization status on sensor.optim_status
 - Bumped setuptools, skforecast, numpy, scipy, pandas
 - A good bunch of documentation improvements thanks to @g1za
+- Improved code coverage (a little bit ;-)
 ### Fix
 - Some fixes managing time zones, thanks to @pail23
 - Bug fix on grid cost function equation, thanks to @michaelpiron

src/emhass/command_line.py

@@ -536,15 +536,15 @@ def publish_data(input_data_dict: dict, logger: logging.Logger,
                                     custom_cost_fun_id["friendly_name"],
                                     type_var = 'cost_fun',
                                     publish_prefix = publish_prefix)
-    # Publish the optimization status (A work in progress, will be available on future release)
-    '''
+    # Publish the optimization status
     custom_cost_fun_id = params['passed_data']['custom_optim_status_id']
-    input_data_dict['rh'].post_data(input_data_dict['opt'].optim_status, idx_closest, 
+    input_data_dict['rh'].post_data(opt_res_latest['optim_status'], idx_closest, 
                                     custom_cost_fun_id["entity_id"], 
                                     custom_cost_fun_id["unit_of_measurement"],
                                     custom_cost_fun_id["friendly_name"],
                                     type_var = 'optim_status',
-                                    publish_prefix = publish_prefix)'''
+                                    publish_prefix = publish_prefix)
+    cols_published = cols_published+["optim_status"]
     # Publish unit_load_cost
     custom_unit_load_cost_id = params['passed_data']['custom_unit_load_cost_id']
     input_data_dict['rh'].post_data(opt_res_latest['unit_load_cost'], idx_closest, 

src/emhass/forecast.py

@@ -316,6 +316,7 @@ def get_weather_forecast(self, method: Optional[str] = 'scrapper',
                 data.set_index('ts', inplace=True)
         else:
             self.logger.error("Method %r is not valid", method)
+            data = None
         return data
 
     def cloud_cover_to_irradiance(self, cloud_cover: pd.Series, 

src/emhass/optimization.py

@@ -461,6 +461,9 @@ def perform_optimization(self, data_opt: pd.DataFrame, P_PV: np.array, P_load: n
                     unit_prod_price[i]*P_grid_neg[i].varValue) for i in set_I]
         else:
             self.logger.error("The cost function specified type is not valid")
+
+        # Add the optimization status
+        opt_tp["optim_status"] = self.optim_status
 
         return opt_tp
 

src/emhass/retrieve_hass.py

@@ -282,6 +282,8 @@ def post_data(self, data_df: pd.DataFrame, idx: int, entity_id: str,
             state = np.round(data_df.sum()[0],2)
         elif type_var == 'unit_load_cost' or type_var == 'unit_prod_price':
             state = np.round(data_df.loc[data_df.index[idx]],4)
+        elif type_var == 'optim_status':
+            state = data_df.loc[data_df.index[idx]]
         else:
             state = np.round(data_df.loc[data_df.index[idx]],2)
         if type_var == 'power':
@@ -305,6 +307,14 @@ def post_data(self, data_df: pd.DataFrame, idx: int, entity_id: str,
         elif type_var == 'mlforecaster':
             data = retrieve_hass.get_attr_data_dict(data_df, idx, entity_id, unit_of_measurement, 
                                                     friendly_name, "scheduled_forecast", state)
+        elif type_var == 'optim_status':
+            data = {
+                "state": state,
+                "attributes": {
+                    "unit_of_measurement": unit_of_measurement,
+                    "friendly_name": friendly_name
+                }
+            }
         else:
             data = {
                 "state": "{:.2f}".format(state),

src/emhass/utils.py

@@ -136,6 +136,7 @@ def treat_runtimeparams(runtimeparams: str, params: str, retrieve_hass_conf: dic
                            'custom_batt_soc_forecast_id': {"entity_id": "sensor.soc_batt_forecast", "unit_of_measurement": "%", "friendly_name": "Battery SOC Forecast"},
                            'custom_grid_forecast_id': {"entity_id": "sensor.p_grid_forecast", "unit_of_measurement": "W", "friendly_name": "Grid Power Forecast"},
                            'custom_cost_fun_id': {"entity_id": "sensor.total_cost_fun_value", "unit_of_measurement": "", "friendly_name": "Total cost function value"},
+                           'custom_optim_status_id': {"entity_id": "sensor.optim_status", "unit_of_measurement": "", "friendly_name": "EMHASS optimization status"},
                            'custom_unit_load_cost_id': {"entity_id": "sensor.unit_load_cost", "unit_of_measurement": "€/kWh", "friendly_name": "Unit Load Cost"},
                            'custom_unit_prod_price_id': {"entity_id": "sensor.unit_prod_price", "unit_of_measurement": "€/kWh", "friendly_name": "Unit Prod Price"},
                            'custom_deferrable_forecast_id': custom_deferrable_forecast_id,
@@ -339,6 +340,8 @@ def treat_runtimeparams(runtimeparams: str, params: str, retrieve_hass_conf: dic
             params['passed_data']['custom_grid_forecast_id'] = runtimeparams['custom_grid_forecast_id']
         if 'custom_cost_fun_id' in runtimeparams.keys():
             params['passed_data']['custom_cost_fun_id'] = runtimeparams['custom_cost_fun_id']
+        if 'custom_optim_status_id' in runtimeparams.keys():
+            params['passed_data']['custom_optim_status_id'] = runtimeparams['custom_optim_status_id']
         if 'custom_unit_load_cost_id' in runtimeparams.keys():
             params['passed_data']['custom_unit_load_cost_id'] = runtimeparams['custom_unit_load_cost_id']
         if 'custom_unit_prod_price_id' in runtimeparams.keys():

tests/test_command_line_utils.py

@@ -234,6 +234,20 @@ def test_naive_mpc_optim(self):
                                               action, logger, get_data_from_file=True)
         opt_res_last = publish_data(input_data_dict, logger, opt_res_latest=opt_res)
         self.assertTrue(len(opt_res_last)==1)
+        # Check if status is published
+        from datetime import datetime
+        now_precise = datetime.now(input_data_dict['retrieve_hass_conf']['time_zone']).replace(second=0, microsecond=0)
+        idx_closest = opt_res.index.get_indexer([now_precise], method='nearest')[0]
+        custom_cost_fun_id = {"entity_id": "sensor.optim_status", "unit_of_measurement": "", "friendly_name": "EMHASS optimization status"}
+        publish_prefix = ""
+        response, data = input_data_dict['rh'].post_data(opt_res['optim_status'], idx_closest, 
+                                        custom_cost_fun_id["entity_id"], 
+                                        custom_cost_fun_id["unit_of_measurement"],
+                                        custom_cost_fun_id["friendly_name"],
+                                        type_var = 'optim_status',
+                                        publish_prefix = publish_prefix)
+        self.assertTrue(hasattr(response, '__class__'))
+        self.assertTrue(data['attributes']['friendly_name'] == 'EMHASS optimization status')
 
     def test_forecast_model_fit_predict_tune(self):
         config_path = pathlib.Path(root+'/config_emhass.yaml')

tests/test_forecast.py

@@ -101,6 +101,8 @@ def test_get_weather_forecast_csv(self):
         self.assertIsInstance(P_PV_forecast.index.dtype, pd.core.dtypes.dtypes.DatetimeTZDtype)
         self.assertEqual(P_PV_forecast.index.tz, self.fcst.time_zone)
         self.assertEqual(len(self.df_weather_csv), len(P_PV_forecast))
+        df_weather_none = self.fcst.get_weather_forecast(method='none')
+        self.assertTrue(df_weather_none == None)
 
     def test_get_weather_forecast_mlforecaster(self):
         pass

tests/test_optimization.py

@@ -79,10 +79,6 @@ def test_perform_dayahead_forecast_optim(self):
         self.assertTrue('cost_fun_'+self.costfun in self.opt_res_dayahead.columns)
         self.assertTrue(self.opt_res_dayahead['P_deferrable0'].sum()*(
             self.retrieve_hass_conf['freq'].seconds/3600) == self.optim_conf['P_deferrable_nom'][0]*self.optim_conf['def_total_hours'][0])
-        # Testing estimation of the current index
-        now_precise = datetime.now(self.input_data_dict['retrieve_hass_conf']['time_zone']).replace(second=0, microsecond=0)
-        idx_closest = self.opt_res_dayahead.index.get_indexer([now_precise], method='ffill')[0]
-        idx_closest = self.opt_res_dayahead.index.get_indexer([now_precise], method='nearest')[0]
         # Test the battery, dynamics and grid exchange contraints
         self.optim_conf.update({'set_use_battery': True})
         self.optim_conf.update({'set_nocharge_from_grid': True})
@@ -103,6 +99,8 @@ def test_perform_dayahead_forecast_optim(self):
         table = opt_res[cost_cols].reset_index().sum(numeric_only=True).to_frame(name='Cost Totals').reset_index()
         self.assertTrue(table.columns[0]=='index')
         self.assertTrue(table.columns[1]=='Cost Totals')
+        # Check status
+        self.assertTrue('optim_status' in self.opt_res_dayahead.columns)
 
     def test_perform_dayahead_forecast_optim_costfun_selfconso(self):
         costfun = 'self-consumption'
@@ -136,7 +134,6 @@ def test_perform_dayahead_forecast_optim_aux(self):
         self.optim_conf['treat_def_as_semi_cont'] = [False, False]
         self.optim_conf['set_total_pv_sell'] = True
         self.optim_conf['set_def_constant'] = [True, True]
-        # self.optim_conf['lp_solver'] = 'GLPK_CMD'
         self.opt = optimization(self.retrieve_hass_conf, self.optim_conf, self.plant_conf, 
                                 self.fcst.var_load_cost, self.fcst.var_prod_price,  
                                 self.costfun, root, logger)
@@ -147,6 +144,20 @@ def test_perform_dayahead_forecast_optim_aux(self):
         self.assertIsInstance(self.opt_res_dayahead, type(pd.DataFrame()))
         self.assertIsInstance(self.opt_res_dayahead.index, pd.core.indexes.datetimes.DatetimeIndex)
         self.assertIsInstance(self.opt_res_dayahead.index.dtype, pd.core.dtypes.dtypes.DatetimeTZDtype)
+        import pulp as pl
+        solver_list = pl.listSolvers(onlyAvailable=True)
+        for solver in solver_list:
+            self.optim_conf['lp_solver'] = solver
+            self.opt = optimization(self.retrieve_hass_conf, self.optim_conf, self.plant_conf, 
+                                    self.fcst.var_load_cost, self.fcst.var_prod_price,  
+                                    self.costfun, root, logger)
+            self.df_input_data_dayahead = self.fcst.get_load_cost_forecast(self.df_input_data_dayahead)
+            self.df_input_data_dayahead = self.fcst.get_prod_price_forecast(self.df_input_data_dayahead)
+            self.opt_res_dayahead = self.opt.perform_dayahead_forecast_optim(
+                self.df_input_data_dayahead, self.P_PV_forecast, self.P_load_forecast)
+            self.assertIsInstance(self.opt_res_dayahead, type(pd.DataFrame()))
+            self.assertIsInstance(self.opt_res_dayahead.index, pd.core.indexes.datetimes.DatetimeIndex)
+            self.assertIsInstance(self.opt_res_dayahead.index.dtype, pd.core.dtypes.dtypes.DatetimeTZDtype)
 
     def test_perform_naive_mpc_optim(self):
         self.df_input_data_dayahead = self.fcst.get_load_cost_forecast(self.df_input_data_dayahead)