Adds support for deferrable load groups

Author: JosephSalisbury

docs/config.md

@@ -103,7 +103,29 @@ Example:
 - `set_deferrable_load_single_constant`: Define if we should set each deferrable load as a constant fixed value variable with just one startup for each optimization task. For example:
 	- False
 	- False
-- `set_deferrable_startup_penalty`: Set to a list of floats. For each deferrable load with a penalty `P`, each time the deferrable load turns on will incur an additional cost of `P * nominal_power_of_deferrable_loads * cost_of_electricity` at that time. 
+- `set_deferrable_startup_penalty`: Set to a list of floats. For each deferrable load with a penalty `P`, each time the deferrable load turns on will incur an additional cost of `P * nominal_power_of_deferrable_loads * cost_of_electricity` at that time.
+- `deferrable_load_groups`: Define groups of deferrable loads that share a physical actuator (e.g. a heat pump serving both hot water and underfloor heating). Each group can enforce a shared power budget, mutual exclusion, or both. This is a list of group objects, each with the following fields:
+	- `names`: List of deferrable load names in the group (e.g. `["deferrable0", "deferrable1"]`).
+	- `max_power` *(optional when `mutual_exclusion` is `true`)*: Maximum combined power in Watts for all loads in the group at any timestep. Required when `mutual_exclusion` is `false`.
+	- `mutual_exclusion` *(optional, defaults to `false`)*: When `true`, only one load in the group may be active at any timestep. Requires all loads in the group to have `treat_deferrable_load_as_semi_cont` set to `true`.
+
+	A load cannot belong to multiple groups. Examples:
+	```json
+	"deferrable_load_groups": [
+	  {"names": ["deferrable0", "deferrable1"], "max_power": 2500}
+	]
+	```
+	```json
+	"deferrable_load_groups": [
+	  {"names": ["deferrable0", "deferrable1"], "mutual_exclusion": true}
+	]
+	```
+	```json
+	"deferrable_load_groups": [
+	  {"names": ["deferrable0", "deferrable1"], "max_power": 2500, "mutual_exclusion": true}
+	]
+	```
+	Defaults to an empty list (no groups).
 - `weather_forecast_method`: This will define the weather forecast method that will be used. The options are `open-meteo` to use the weather forecast API proposed by [Open-Meteo](https://open-meteo.com/), `solcast` to use the [Solcast](https://solcast.com/) solar forecast service, `solar.forecast` to use the free public [Solar.Forecast](https://forecast.solar/) account and finally the `csv` to load a CSV file. When loading a CSV file this will be directly considered as the PV power forecast in Watts. The default CSV file path that will be used is `/data/data_weather_forecast.csv`. This method is useful to load and use other external forecasting service data in EMHASS. Defaults to `open-meteo` method.
 - `load_forecast_method`: The load forecast method that will be used. The options are `typical` which uses basic statistics and a year long load power data grouped by the current day-of-the-week of the current month, `naive` also called persistence that assumes that the forecast for a future period will be equal to the observed values in a past period, `mlforecaster` that uses regression models considering auto-regression lags as features and finally the `csv` to load a CSV file. When loading a CSV file this will be directly considered as the PV power forecast in Watts. The default CSV file path that will be used is `/data/data_weather_forecast.csv`. This method is useful to load and use other external forecasting service data in EMHASS. Defaults to `typical`.
 ```{note} 

src/emhass/command_line.py

@@ -122,6 +122,7 @@ class OptimizationCacheKey:
     set_deferrable_load_as_timeseries: tuple
     nominal_power_of_deferrable_loads: tuple
     def_load_config_structure: tuple  # (index, type) tuples for each load
+    deferrable_load_groups: tuple
     inverter_is_hybrid: bool
     compute_curtailment: bool
     optimization_time_step_s: float | None
@@ -284,6 +285,10 @@ def config_hash(cfg: dict, exclude_keys: set | None = None) -> str:
                 optim_conf.get("nominal_power_of_deferrable_loads", [])
             ),
             def_load_config_structure=tuple(def_structure),
+            deferrable_load_groups=tuple(
+                (tuple(g.get("names", [])), g.get("max_power"), g.get("mutual_exclusion", False))
+                for g in optim_conf.get("deferrable_load_groups", [])
+            ),
             inverter_is_hybrid=plant_conf.get("inverter_is_hybrid", False),
             compute_curtailment=plant_conf.get("compute_curtailment", False),
             optimization_time_step_s=to_seconds(retrieve_hass_conf.get("optimization_time_step")),

src/emhass/data/associations.csv

@@ -69,6 +69,7 @@ optim_conf,open_meteo_cache_max_age,open_meteo_cache_max_age
 optim_conf,def_start_timestep,start_timesteps_of_each_deferrable_load,list_start_timesteps_of_each_deferrable_load
 optim_conf,def_end_timestep,end_timesteps_of_each_deferrable_load,list_end_timesteps_of_each_deferrable_load
 optim_conf,set_deferrable_max_startups,set_deferrable_max_startups
+optim_conf,deferrable_load_groups,deferrable_load_groups
 optim_conf,list_hp_periods,load_peak_hour_periods
 optim_conf,model_type,model_type
 optim_conf,var_model,var_model

src/emhass/data/config_defaults.json

@@ -76,6 +76,7 @@
   "set_deferrable_load_single_constant": [false, false],
   "set_deferrable_startup_penalty": [0.0, 0.0],
   "set_deferrable_max_startups": [0, 0],
+  "deferrable_load_groups": [],
   "load_peak_hours_cost": 0.1907,
   "load_offpeak_hours_cost": 0.1419,
   "production_price_forecast_method": "constant",

src/emhass/optimization.py

@@ -2004,6 +2004,39 @@ def _add_deferrable_load_constraints(
 
         return predicted_temps, heating_demands, penalty_terms_total, q_inputs
 
+    def _add_deferrable_group_constraints(self, constraints, relaxed=False):
+        """Add shared power budget and mutual exclusion constraints for deferrable load groups.
+
+        Args:
+            constraints: List of CVXPY constraints to append to.
+            relaxed: If True, only add shared power budget constraints (skip mutual
+                exclusion, which requires binary variables not available in the relaxed LP).
+        """
+        groups = self.optim_conf.get("deferrable_load_groups", [])
+        if not groups:
+            return
+
+        p_deferrable = self.vars["p_deferrable"]
+
+        for gi, group in enumerate(groups):
+            indices = [int(name.replace("deferrable", "")) for name in group["names"]]
+            max_power = group.get("max_power")
+            mutual_exclusion = group.get("mutual_exclusion", False)
+
+            self.logger.debug(f"Adding group {gi} constraints for deferrable loads {indices}")
+
+            # Shared power budget: sum of group members <= max_power at each timestep
+            if max_power is not None:
+                group_power_sum = sum(p_deferrable[i] for i in indices)
+                constraints.append(group_power_sum <= max_power)
+
+            # Mutual exclusion: at most one load active per timestep
+            # Skipped in relaxed mode as binary variables are not available
+            if mutual_exclusion and not relaxed:
+                p_def_bin2 = self.vars["p_def_bin2"]
+                bin_sum = sum(p_def_bin2[i] for i in indices)
+                constraints.append(bin_sum <= 1)
+
     def _build_results_dataframe(
         self,
         data_opt,
@@ -2457,6 +2490,9 @@ def pad_list(input_list, target_len, fill=0):
                 )
             )
 
+            # Deferrable Load Group Constraints (shared power budget, mutual exclusion)
+            self._add_deferrable_group_constraints(constraints)
+
             # Build Objective
             objective_expr = self._build_objective_function(
                 batt_stress_conf,
@@ -2599,6 +2635,9 @@ def pad_list(input_list, target_len, fill=0):
                 )
             )
 
+            # Deferrable Load Group Constraints (shared power budget only in relaxed mode)
+            self._add_deferrable_group_constraints(constraints_relaxed, relaxed=True)
+
             # Re-build Objective
             objective_expr = self._build_objective_function(batt_stress_conf, inv_stress_conf)
             if not isinstance(penalty_terms_total, int) or penalty_terms_total != 0:

src/emhass/utils.py

@@ -2182,6 +2182,68 @@ async def build_params(
             "nominal_power_of_deferrable_loads",
             logger,
         )
+        # Validate deferrable_load_groups
+        groups = params["optim_conf"].get("deferrable_load_groups", [])
+        if groups:
+            seen_indices = set()
+            for gi, group in enumerate(groups):
+                # Validate names
+                names = group.get("names", [])
+                if not names:
+                    raise ValueError(
+                        f"deferrable_load_groups[{gi}]: 'names' must contain at least 1 deferrable load reference"
+                    )
+                indices = []
+                group_indices = set()
+                for name in names:
+                    try:
+                        idx = int(name.replace("deferrable", ""))
+                    except ValueError:
+                        raise ValueError(
+                            f"deferrable_load_groups[{gi}]: could not parse index from name '{name}'"
+                        )
+                    if idx < 0 or idx >= num_def_loads:
+                        raise ValueError(
+                            f"deferrable_load_groups[{gi}]: '{name}' references index {idx}, "
+                            f"but only {num_def_loads} deferrable loads are configured"
+                        )
+                    if idx in group_indices:
+                        raise ValueError(
+                            f"deferrable_load_groups[{gi}]: '{name}' is duplicated within the group"
+                        )
+                    if idx in seen_indices:
+                        raise ValueError(
+                            f"deferrable_load_groups[{gi}]: '{name}' is already in another group. "
+                            f"A deferrable load cannot belong to multiple groups"
+                        )
+                    indices.append(idx)
+                    group_indices.add(idx)
+                seen_indices.update(indices)
+                
+                # Validate max_power (optional when mutual_exclusion is true)
+                max_power = group.get("max_power")
+                mutual_exclusion = group.get("mutual_exclusion", False)
+                if max_power is not None and max_power <= 0:
+                    raise ValueError(
+                        f"deferrable_load_groups[{gi}]: 'max_power' must be a positive number"
+                    )
+                if not isinstance(mutual_exclusion, bool):
+                    raise ValueError(
+                        f"deferrable_load_groups[{gi}]: 'mutual_exclusion' must be a boolean"
+                    )
+                if max_power is None and not mutual_exclusion:
+                    raise ValueError(
+                        f"deferrable_load_groups[{gi}]: 'max_power' is required when 'mutual_exclusion' is false"
+                    )
+                if mutual_exclusion:
+                    semi_cont = params["optim_conf"].get("treat_deferrable_load_as_semi_cont", [])
+                    for idx in indices:
+                        is_semi_cont = semi_cont[idx] if idx < len(semi_cont) else False
+                        if not is_semi_cont:
+                            raise ValueError(
+                                f"deferrable_load_groups[{gi}]: mutual_exclusion requires "
+                                f"'deferrable{idx}' to have treat_deferrable_load_as_semi_cont=true"
+                            )
     else:
         logger.warning("unable to obtain parameter: number_of_deferrable_loads")
     # historic_days_to_retrieve should be no less then 2

tests/test_optimization.py

@@ -3889,6 +3889,130 @@ def test_load_deactivation_with_def_total_hours(self):
         )
 
 
+    def test_deferrable_load_group_shared_power(self):
+        """Test that shared power budget constraint limits combined power of grouped loads."""
+        self.optim_conf.update(
+            {
+                "treat_deferrable_load_as_semi_cont": [True, True],
+                "set_deferrable_load_single_constant": [False, False],
+                "nominal_power_of_deferrable_loads": [2000.0, 2000.0],
+                "operating_hours_of_each_deferrable_load": [4, 4],
+                "deferrable_load_groups": [
+                    {
+                        "names": ["deferrable0", "deferrable1"],
+                        "max_power": 2500,
+                        "mutual_exclusion": False,
+                    }
+                ],
+            }
+        )
+        self.opt = self.create_optimization()
+        self.df_input_data_dayahead = self.prepare_forecast_data()
+        opt_res = self.opt.perform_dayahead_forecast_optim(
+            self.df_input_data_dayahead, self.p_pv_forecast, self.p_load_forecast
+        )
+        self.assertIn(self.opt.optim_status, VALID_OPTIMAL_STATUSES)
+        # Verify combined power never exceeds group max_power (with small tolerance)
+        combined = opt_res["P_deferrable0"] + opt_res["P_deferrable1"]
+        self.assertTrue(
+            (combined <= 2500 + 1.0).all(),
+            f"Combined power exceeded group max_power: max={combined.max():.1f}",
+        )
+
+    def test_deferrable_load_group_mutual_exclusion(self):
+        """Test that mutual exclusion prevents simultaneous operation of grouped loads."""
+        self.optim_conf.update(
+            {
+                "treat_deferrable_load_as_semi_cont": [True, True],
+                "set_deferrable_load_single_constant": [False, False],
+                "nominal_power_of_deferrable_loads": [2000.0, 1500.0],
+                "operating_hours_of_each_deferrable_load": [4, 4],
+                "deferrable_load_groups": [
+                    {
+                        "names": ["deferrable0", "deferrable1"],
+                        "max_power": 2500,
+                        "mutual_exclusion": True,
+                    }
+                ],
+            }
+        )
+        self.opt = self.create_optimization()
+        self.df_input_data_dayahead = self.prepare_forecast_data()
+        opt_res = self.opt.perform_dayahead_forecast_optim(
+            self.df_input_data_dayahead, self.p_pv_forecast, self.p_load_forecast
+        )
+        self.assertIn(self.opt.optim_status, VALID_OPTIMAL_STATUSES)
+        # Verify at most one load is active at any timestep
+        both_active = (opt_res["P_deferrable0"] > 1.0) & (opt_res["P_deferrable1"] > 1.0)
+        self.assertFalse(
+            both_active.any(),
+            "Mutual exclusion violated: both loads active simultaneously",
+        )
+
+    def test_deferrable_load_group_no_groups(self):
+        """Test that empty deferrable_load_groups works (backward compatibility)."""
+        self.optim_conf["deferrable_load_groups"] = []
+        self.opt = self.create_optimization()
+        self.df_input_data_dayahead = self.prepare_forecast_data()
+        opt_res = self.opt.perform_dayahead_forecast_optim(
+            self.df_input_data_dayahead, self.p_pv_forecast, self.p_load_forecast
+        )
+        self.assertIn(self.opt.optim_status, VALID_OPTIMAL_STATUSES)
+
+    async def _build_params_with_groups(self, groups, **config_overrides):
+        """Helper to build params with deferrable_load_groups set in config."""
+        config = await build_config(emhass_conf, logger, emhass_conf["defaults_path"])
+        config["deferrable_load_groups"] = groups
+        for key, value in config_overrides.items():
+            config[key] = value
+        _, secrets = await build_secrets(emhass_conf, logger, no_response=True)
+        return await build_params(emhass_conf, secrets, config, logger)
+
+    async def test_deferrable_load_group_validation_invalid_name(self):
+        """Test that invalid deferrable names in groups raise errors."""
+        with self.assertRaises(ValueError):
+            await self._build_params_with_groups([
+                {
+                    "names": ["deferrable0", "deferrable99"],
+                    "max_power": 2500,
+                    "mutual_exclusion": False,
+                }
+            ])
+
+    async def test_deferrable_load_group_validation_mutual_exclusion_not_semi_cont(self):
+        """Test that mutual exclusion with non-semi-continuous loads raises error."""
+        with self.assertRaises(ValueError):
+            await self._build_params_with_groups(
+                [
+                    {
+                        "names": ["deferrable0", "deferrable1"],
+                        "max_power": 2500,
+                        "mutual_exclusion": True,
+                    }
+                ],
+                treat_deferrable_load_as_semi_cont=[False, False],
+            )
+
+    async def test_deferrable_load_group_validation_overlapping_groups(self):
+        """Test that a load in multiple groups raises error."""
+        with self.assertRaises(ValueError):
+            await self._build_params_with_groups(
+                [
+                    {
+                        "names": ["deferrable0", "deferrable1"],
+                        "max_power": 2500,
+                        "mutual_exclusion": False,
+                    },
+                    {
+                        "names": ["deferrable1", "deferrable2"],
+                        "max_power": 2000,
+                        "mutual_exclusion": False,
+                    },
+                ],
+                number_of_deferrable_loads=3,
+            )
+
+
 if __name__ == "__main__":
     unittest.main()
     ch.close()