Add dynamic additional load forecasts

apps/predbat/config.py

@@ -1133,6 +1133,16 @@
         "restore": False,
         "api": True,
     },
+    {
+        "name": "load_forecast_delta_api",
+        "friendly_name": "Load forecast delta API controls",
+        "type": "select",
+        "options": ["off"],
+        "icon": "mdi:dishwasher",
+        "default": "off",
+        "restore": False,
+        "api": True,
+    },
     {
         "name": "manual_freeze_charge",
         "friendly_name": "Manual force charge freeze",
@@ -2043,6 +2053,7 @@
         "type": "sensor_list",
         "sensor_type": "dict|list",
     },
+    "house_load_additional_forecast": {"type": "dict_list"},
     "ge_cloud_data": {"type": "boolean"},
     "ge_cloud_serial": {"type": "string", "empty": False},
     "ge_cloud_key": {"type": "string", "empty": False},

apps/predbat/ha.py

@@ -890,13 +890,23 @@ def call_service(self, service, **kwargs):
             data_frame = {"domain": domain, "service": service, "service_data": data}
             return run_async(self.base.trigger_callback(data_frame))
 
-    def api_call(self, endpoint, data_in=None, post=False, core=True, silent=False):
+    def delete_state(self, entity_id):
+        """
+        Delete a state from Home Assistant.
+        """
+        self.db_mirror_list.pop(entity_id, None)
+        self.state_data.pop(entity_id.lower(), None)
+        if self.ha_key:
+            self.api_call("/api/states/{}".format(entity_id), delete=True)
+
+    def api_call(self, endpoint, data_in=None, post=False, delete=False, core=True, silent=False):
         """
         Make an API call to Home Assistant.
 
         :param endpoint: The API endpoint to call.
         :param data_in: The data to send in the body of the request.
         :param post: True if this is a POST request, False for GET.
+        :param delete: True if this is a DELETE request
         :param core: True is this is a call to HA Core, False if it is a Supervisor call
         :param silent: True if warning message from the API call is to be suppressed
         :return: The response from the API.
@@ -918,7 +928,9 @@ def api_call(self, endpoint, data_in=None, post=False, core=True, silent=False):
             "Accept": "application/json",
         }
         try:
-            if post:
+            if delete:
+                response = requests.delete(url, headers=headers, timeout=TIMEOUT)
+            elif post:
                 if data_in:
                     response = requests.post(url, headers=headers, json=data_in, timeout=TIMEOUT)
                 else:
@@ -928,7 +940,7 @@ def api_call(self, endpoint, data_in=None, post=False, core=True, silent=False):
                     response = requests.get(url, headers=headers, params=data_in, timeout=TIMEOUT)
                 else:
                     response = requests.get(url, headers=headers, timeout=TIMEOUT)
-            data = response.json()
+            data = {} if delete and not response.text else response.json()
             self.api_errors = 0
         except requests.exceptions.JSONDecodeError:
             if not silent:  # suppress warning message for call to get slug id from supervisor because in docker installs this will always error (no supervisor)

apps/predbat/output.py

@@ -32,6 +32,43 @@ class Output:
     charging schedules, and financial metric summaries.
     """
 
+    def additional_load_plan_time(self, timestamp):
+        """
+        Return a compact local time string for an additional load timestamp.
+        """
+        return datetime.fromisoformat(timestamp).strftime("%H:%M")
+
+    def get_additional_load_text(self):
+        """
+        Return a textual summary of confirmed planned additional load forecasts.
+        """
+        planned_loads = []
+        for name, forecast in sorted(getattr(self, "house_load_additional_forecasts", {}).items()):
+            target_times = forecast.get("target_times", [])
+            total_energy = forecast.get("total_energy", 0.0)
+            if not forecast.get("enabled", False) or not target_times or total_energy <= 0:
+                continue
+            start = target_times[0].get("start")
+            end = target_times[-1].get("end")
+            if not start or not end:
+                continue
+            planned_loads.append(
+                {
+                    "name": name,
+                    "start": start,
+                    "end": end,
+                    "text": "{} from {} to {} using {:.2f} kWh".format(name, self.additional_load_plan_time(start), self.additional_load_plan_time(end), dp2(total_energy)),
+                }
+            )
+
+        if not planned_loads:
+            return ""
+
+        planned_loads = sorted(planned_loads, key=lambda load: load["start"])
+        if len(planned_loads) == 1:
+            return "- Additional load {} is planned.\n".format(planned_loads[0]["text"])
+        return "- Additional loads are planned: {}.\n".format("; ".join(load["text"] for load in planned_loads))
+
     def publish_car_plan(self):
         """
         Publish the car charging plan
@@ -915,6 +952,8 @@ def short_textual_plan(self, soc_min, soc_min_minute, pv_forecast_minute_step, p
         if car_charging_kwh > 0:
             sentence += "- Your car is currently charging.\n"
 
+        sentence += self.get_additional_load_text()
+
         charge_window_n_next = self.get_next_charge_window(self.minutes_now)
         export_window_n_next = self.get_next_export_window(self.minutes_now)
         if charge_window_n < 0 and charge_window_n_next >= 0:

apps/predbat/plan.py

@@ -67,6 +67,120 @@ class Plan:
     runs to minimise the overall cost metric.
     """
 
+    def additional_load_candidate_profile(self, forecast, start_minutes):
+        """
+        Build absolute-minute adjustment and target metadata for one flexible load candidate.
+        """
+        plan_interval = forecast.get("plan_interval_minutes", self.plan_interval_minutes)
+        duration_minutes = int(forecast.get("duration", 0.0) * 60)
+        end_minutes = start_minutes + duration_minutes
+        periods = forecast.get("_periods", 0)
+        weights = forecast.get("_weights", [])
+        weight_total = forecast.get("_weight_total", sum(weights))
+        energy_total = forecast.get("energy", None)
+        slot_energy = forecast.get("slot_energy", 0.0)
+        load_adjust = {}
+        target_times = []
+        total_energy = 0.0
+
+        for period in range(periods):
+            slot_start = start_minutes + period * plan_interval
+            slot_end = min(slot_start + plan_interval, end_minutes)
+            slot_minutes = slot_end - slot_start
+            if slot_end <= self.minutes_now:
+                continue
+            if (slot_start - self.minutes_now) >= self.forecast_minutes:
+                continue
+            energy, adjustment_energy = self.additional_load_slot_energies(energy_total, slot_energy, weights, weight_total, period, slot_minutes, plan_interval)
+            total_energy += energy
+            for minute in range(slot_start, slot_end):
+                load_adjust[minute] = dp4(load_adjust.get(minute, 0.0) + adjustment_energy)
+            target_times.append({"start": (self.midnight_utc + timedelta(minutes=slot_start)).isoformat(), "end": (self.midnight_utc + timedelta(minutes=slot_end)).isoformat(), "energy": energy})
+
+        return load_adjust, target_times, dp4(total_energy)
+
+    def add_additional_load_to_step_data(self, load_minutes_step, load_adjust):
+        """
+        Add absolute-minute additional load adjustment into prediction step data.
+        """
+        modified_load = copy.deepcopy(load_minutes_step)
+        for minute_absolute, energy in load_adjust.items():
+            minute_relative = minute_absolute - self.minutes_now
+            if minute_relative < 0 or minute_relative >= self.forecast_minutes:
+                continue
+            step_minute = int(minute_relative / PREDICT_STEP) * PREDICT_STEP
+            modified_load[step_minute] = dp4(modified_load.get(step_minute, 0.0) + energy * PREDICT_STEP / float(self.plan_interval_minutes))
+        return modified_load
+
+    def select_flexible_additional_loads(self, load_minutes_step, load_minutes_step10, pv_forecast_minute_step, pv_forecast_minute10_step):
+        """
+        Select flexible additional load start times using full prediction metric impact.
+        """
+        flexible_forecasts = {name: forecast for name, forecast in self.house_load_additional_forecasts.items() if forecast.get("enabled") and forecast.get("mode") == "flexible" and not forecast.get("selection_locked", False)}
+        if not flexible_forecasts:
+            return False, load_minutes_step, load_minutes_step10
+
+        selected_flexible = {}
+        working_load_step = load_minutes_step
+        working_load_step10 = load_minutes_step10
+
+        for name, forecast in flexible_forecasts.items():
+            start_minutes = forecast.get("_requested_start_minutes", None)
+            end_minutes = forecast.get("_requested_end_minutes", None)
+            duration_minutes = int(forecast.get("duration", 0.0) * 60)
+            plan_interval = forecast.get("plan_interval_minutes", self.plan_interval_minutes)
+            if start_minutes is None or end_minutes is None or duration_minutes <= 0:
+                continue
+
+            candidate = max(start_minutes, self.minutes_now)
+            candidate = int((candidate + plan_interval - 1) / plan_interval) * plan_interval
+            latest_start = min(end_minutes - duration_minutes, self.minutes_now + self.forecast_minutes - duration_minutes)
+            if latest_start < candidate:
+                continue
+
+            baseline_prediction = Prediction(self, pv_forecast_minute_step, pv_forecast_minute10_step, working_load_step, working_load_step10)
+            baseline_metric = baseline_prediction.run_prediction(self.charge_limit_best, self.charge_window_best, self.export_window_best, self.export_limits_best, False, self.end_record)[0]
+            best_start = None
+            best_metric = None
+            candidate_count = 0
+
+            while candidate <= latest_start:
+                candidate_adjust, _, _ = self.additional_load_candidate_profile(forecast, candidate)
+                candidate_load_step = self.add_additional_load_to_step_data(working_load_step, candidate_adjust)
+                candidate_load_step10 = self.add_additional_load_to_step_data(working_load_step10, candidate_adjust)
+                candidate_prediction = Prediction(self, pv_forecast_minute_step, pv_forecast_minute10_step, candidate_load_step, candidate_load_step10)
+                candidate_metric = candidate_prediction.run_prediction(self.charge_limit_best, self.charge_window_best, self.export_window_best, self.export_limits_best, False, self.end_record)[0]
+                candidate_count += 1
+                if best_metric is None or candidate_metric < best_metric:
+                    best_metric = candidate_metric
+                    best_start = candidate
+                candidate += plan_interval
+
+            if best_start is not None:
+                best_adjust, _, _ = self.additional_load_candidate_profile(forecast, best_start)
+                working_load_step = self.add_additional_load_to_step_data(working_load_step, best_adjust)
+                working_load_step10 = self.add_additional_load_to_step_data(working_load_step10, best_adjust)
+                selected_flexible[name] = {
+                    "_requested_start_minutes": start_minutes,
+                    "_requested_end_minutes": end_minutes,
+                    "_selected_start_minutes": best_start,
+                    "_selection_reason": "prediction_metric",
+                    "_candidate_count": candidate_count,
+                    "_selected_metric": dp2(best_metric) if best_metric is not None else None,
+                    "_baseline_metric": dp2(baseline_metric),
+                    "_expires_minutes": best_start + duration_minutes if forecast.get("auto_expire", False) else None,
+                }
+                if forecast.get("auto_expire", False):
+                    self.house_load_additional_forecast_overrides[name] = {"name": name, **selected_flexible[name]}
+                self.log("Flexible additional load {} selected {}-{} using prediction metric {} from {} candidates".format(name, self.time_abs_str(best_start), self.time_abs_str(best_start + duration_minutes), dp2(best_metric), candidate_count))
+
+        if not selected_flexible:
+            return False, load_minutes_step, load_minutes_step10
+
+        self.house_load_additional_forecast_adjust, self.house_load_additional_forecasts = self.fetch_additional_load_forecast(selected_flexible=selected_flexible)
+        self.publish_additional_load_forecasts()
+        return True, working_load_step, working_load_step10
+
     def dynamic_load(self):
         """
         Adjust load prediction based on current load
@@ -930,6 +1044,9 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
         # Created optimised step data
         self.metric_cloud_coverage = self.get_cloud_factor(self.minutes_now, self.pv_forecast_minute, self.pv_forecast_minute10)
         self.metric_load_divergence = self.get_load_divergence(self.minutes_now, self.load_minutes)
+        load_adjust = self.manual_load_adjust.copy()
+        for minute, adjustment in self.house_load_additional_forecast_adjust.items():
+            load_adjust[minute] = load_adjust.get(minute, 0.0) + adjustment
         load_minutes_step = self.step_data_history(
             self.load_minutes,
             self.minutes_now,
@@ -940,7 +1057,7 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
             load_forecast=self.load_forecast,
             load_scaling_dynamic=self.load_scaling_dynamic,
             cloud_factor=self.metric_load_divergence,
-            load_adjust=self.manual_load_adjust,
+            load_adjust=load_adjust,
             load_baseline=self.dynamic_load_baseline,
         )
         load_minutes_step10 = self.step_data_history(
@@ -953,7 +1070,7 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
             load_forecast=self.load_forecast,
             load_scaling_dynamic=self.load_scaling_dynamic,
             cloud_factor=min(self.metric_load_divergence + 0.5, 1.0) if self.metric_load_divergence else None,
-            load_adjust=self.manual_load_adjust,
+            load_adjust=load_adjust,
             load_baseline=self.dynamic_load_baseline,
         )
         pv_forecast_minute_step = self.step_data_history(self.pv_forecast_minute, self.minutes_now, forward=True, cloud_factor=self.metric_cloud_coverage)
@@ -972,6 +1089,12 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
         # Creation prediction object
         self.prediction = Prediction(self, pv_forecast_minute_step, pv_forecast_minute10_step, load_minutes_step, load_minutes_step10)
 
+        flexible_selected, load_minutes_step, load_minutes_step10 = self.select_flexible_additional_loads(load_minutes_step, load_minutes_step10, pv_forecast_minute_step, pv_forecast_minute10_step)
+        if flexible_selected:
+            self.load_minutes_step = load_minutes_step
+            self.load_minutes_step10 = load_minutes_step10
+            self.prediction = Prediction(self, pv_forecast_minute_step, pv_forecast_minute10_step, load_minutes_step, load_minutes_step10)
+
         # Check if LoadML is active and disable thread pools as it causes lockup due to race conditions with NumPy
         load_ml_comp = self.components.get_component("load_ml") if self.components else None
         load_ml_calculating = False

apps/predbat/predbat.py

@@ -333,6 +333,18 @@ def set_state_wrapper(self, entity_id, state, attributes={}, required_unit=None)
         state = self.unit_conversion(entity_id, state, None, required_unit, going_to=True)
         return self.ha_interface.set_state(entity_id, state, attributes=attributes)
 
+    def delete_state_wrapper(self, entity_id):
+        """
+        Wrapper function to delete state from HA.
+        """
+        if not self.ha_interface:
+            self.log("Error: delete_state_wrapper - No HA interface available")
+            return False
+        if not hasattr(self.ha_interface, "delete_state"):
+            return False
+
+        return self.ha_interface.delete_state(entity_id)
+
     def fire_event_wrapper(self, domain, service):
         """
         Wrapper function to fire a HA event
@@ -462,9 +474,13 @@ def reset(self):
         self.manual_demand_times = []
         self.manual_all_times = []
         self.manual_api = []
+        self.load_forecast_delta_api = []
         self.manual_import_rates = {}
         self.manual_export_rates = {}
         self.manual_load_adjust = {}
+        self.house_load_additional_forecast_adjust = {}
+        self.house_load_additional_forecasts = {}
+        self.house_load_additional_forecast_overrides = {}
         self.config_index = {}
         self.dashboard_index = []
         self.dashboard_index_app = {}