Add configurable whole-site import diagnostics (no enforcement)

apps/predbat/config.py

@@ -2712,6 +2712,8 @@
     "inverter_limit_override": {"type": "sensor_list", "sensor_type": "integer", "modify": False, "zero": False, "entries": "num_inverters"},
     "battery_rate_max": {"type": "sensor_list", "sensor_type": "float", "modify": False, "zero": False, "entries": "num_inverters"},
     "export_limit": {"type": "sensor_list", "sensor_type": "float", "entries": "num_inverters"},
+    # Whole-site diagnostic threshold in W, not a per-inverter limit or a control setting.
+    "import_limit": {"type": "float", "zero": True},
     "inverter_battery_rate_min": {"type": "integer", "zero": False, "entries": "num_inverters"},
     "inverter_reserve_max": {"type": "integer", "zero": False, "entries": "num_inverters"},
     "battery_charge_power_curve": {"type": "int_float_dict", "or_auto": True},

apps/predbat/const.py

@@ -101,6 +101,8 @@
 INVERTER_QUICK_UPDATE_SECONDS = 60  # Minimum seconds between quick inverter data updates, also the balance re-apply interval
 INVERTER_LIMIT_DEFAULT_W = 7500.0  # Standing inverter AC limit when the user has not configured inverter_limit
 EXPORT_LIMIT_DEFAULT_W = 99999.0  # Standing export limit when the user has not configured export_limit - effectively unlimited
+# Whole-site diagnostic threshold in W; 0 disables reporting. Not a control or optimiser constraint.
+IMPORT_LIMIT_DEFAULT_W = 0.0
 PREDBAT_MAX_CARS = 8  # Matches PK_MAX_CARS in prediction_kernel.cpp and the car_charging_rate/_1../_7 config items - the hard ceiling on num_cars
 # A car_charging_now sensor that reports a charging power (in W, after unit conversion) rather than
 # on/off counts as charging from this power: above a charger's standby draw, below the smallest charge

apps/predbat/fetch.py

@@ -19,6 +19,7 @@
 """
 
 from datetime import datetime, timedelta
+from math import isfinite
 from utils import minutes_to_time, str2time, dp1, dp2, dp3, dp4, time_string_to_stamp, minute_data, get_now_from_cumulative, MinuteArray
 from const import (
     MINUTE_WATT,
@@ -36,6 +37,7 @@
     CLOUD_WINDOW_MINUTES,
     CLOUD_ARRAY_MARGIN,
     PV_ARRAY_KWP_UNKNOWN,
+    IMPORT_LIMIT_DEFAULT_W,
 )
 from predbat_metrics import metrics
 from futurerate import FutureRate
@@ -82,6 +84,18 @@ class Fetch:
     and other sensor inputs. Converts all data to per-minute dictionaries.
     """
 
+    def fetch_import_limit(self):
+        """Read one finite, non-negative site threshold in W; never coerce bad input to off.
+
+        This first stage accepts a literal YAML number only. Read the raw argument because
+        get_arg with a numeric default coerces invalid values to that default, hiding errors.
+        The generic float schema does not reject lists, booleans or non-finite values.
+        """
+        value = self.args.get("import_limit", IMPORT_LIMIT_DEFAULT_W)
+        if type(value) not in (int, float) or not isfinite(value) or value < 0:
+            raise ValueError("import_limit must be a finite non-negative scalar number in W (0 disables diagnostics)")
+        return float(value)
+
     def get_cloud_factor(self, minutes_now, pv_data, pv_data10):
         """
         Work out approximated cloud factor
@@ -3746,6 +3760,7 @@ def fetch_config_options(self):
         self.inverter_loss = 1.0 - self.get_arg("inverter_loss")
         self.inverter_hybrid = self.get_arg("inverter_hybrid")
         self.pv_ac_limit = self.get_arg("pv_ac_limit", 0.0) / MINUTE_WATT
+        self.import_limit = self.fetch_import_limit()
         self.inverter_freeze_export_discharge_rate = max(self.get_arg("inverter_freeze_export_discharge_rate", 0.0), 0.0) / MINUTE_WATT
         self.log("Freeze Export discharge rate configured: {:.0f} W".format(self.inverter_freeze_export_discharge_rate * MINUTE_WATT))
         self.base_load = self.get_arg("base_load", 100) / 1000.0

apps/predbat/plan.py

@@ -5105,6 +5105,32 @@ def optimise_charge_windows_reset(self, reset_all):
                 else:
                     self.export_limits_best[window_n] = pack_export_limit(EXPORT_MODE_IDLE)
 
+    def publish_import_limit_diagnostics(self, pred):
+        """Publish the final best forecast's diagnostic, never an enforcement/feasibility claim."""
+        attributes = {
+            "friendly_name": "Best plan grid import limit diagnostic",
+            "icon": "mdi:transmission-tower-import",
+            "scope": "whole_site",
+            "enforced": False,
+            "limit_w": self.import_limit,
+            "peak_w": None,
+            "excess_kwh": None,
+            "minutes_over": None,
+            "first_exceeded": None,
+            "step_minutes": None,
+            "checked_minutes": 0,
+            "start": None,
+            "end": None,
+        }
+        state = "disabled" if self.import_limit == 0 else "unavailable"
+        diagnostic = getattr(pred, "import_limit_diagnostics", None)
+        if self.import_limit > 0 and diagnostic and diagnostic["limit_w"] == self.import_limit and diagnostic["checked_minutes"] == pred.forecast_minutes and pred.forecast_minutes > 0:
+            attributes.update(diagnostic)
+            attributes["peak_w"] = round(diagnostic["peak_w"], 3)
+            attributes["excess_kwh"] = round(diagnostic["excess_kwh"], 6)
+            state = "exceeded" if diagnostic["minutes_over"] else "within_limit"
+        self.dashboard_item(self.prefix + ".best_import_limit", state=state, attributes=attributes)
+
     def run_prediction(self, charge_limit, charge_window, export_window, export_limits, pv_scenario, end_record, save=None, step=PREDICT_STEP):
         """
         Run a prediction scenario given a charge limit, options to save the results or not to HA entity
@@ -5400,6 +5426,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
 
             if save and save == "best":
                 self.log("Saving plan best values to HA entities")
+                self.publish_import_limit_diagnostics(pred)
                 self.dashboard_item(
                     self.prefix + ".best_battery_hours_left",
                     state=dp2(hours_left),

apps/predbat/predbat.py

@@ -503,6 +503,8 @@ def reset(self):
         self.current_charge_limit_kwh = 0.0
         self.inverter_limit = 0.0
         self.export_limit = 0.0
+        # Whole-site diagnostic threshold, stored in W (not kWh/minute). 0 disables reporting.
+        self.import_limit = 0.0
         self.charge_limit = []
         self.charge_limit_best = []
         self.charge_window = []

apps/predbat/prediction.py

@@ -18,7 +18,7 @@
 """
 
 from datetime import timedelta
-from const import PREDICT_STEP, PV_SCENARIO_PV10, PV_SCENARIO_PV90, RUN_EVERY, TIME_FORMAT, EXPORT_LIMIT_FREEZE, EXPORT_LIMIT_IDLE, EXPORT_MODE_TARGET, EXPORT_MODE_FREEZE, EXPORT_MODE_IDLE
+from const import PREDICT_STEP, PV_SCENARIO_PV10, PV_SCENARIO_PV90, RUN_EVERY, TIME_FORMAT, EXPORT_LIMIT_FREEZE, EXPORT_LIMIT_IDLE, EXPORT_MODE_TARGET, EXPORT_MODE_FREEZE, EXPORT_MODE_IDLE, MINUTE_WATT
 
 from utils import (
     remove_intersecting_windows,
@@ -172,6 +172,8 @@ def __init__(
             self.inverter_hybrid = base.inverter_hybrid
             self.inverter_limit = base.inverter_limit
             self.export_limit = base.export_limit
+            # Diagnostic only, in W. It does not modify dispatch, costs or candidate selection.
+            self.import_limit = getattr(base, "import_limit", 0.0)
             self.pv_ac_limit = base.pv_ac_limit
             self.battery_rate_min = base.battery_rate_min
             self.battery_rate_max_charge = base.battery_rate_max_charge
@@ -520,6 +522,10 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
         if save and self.pending_batch:
             self.flush_batch()
 
+        # Only saved Python runs report this diagnostic. Clear it even on a cache/kernel return
+        # so a later trial cannot appear to have inherited a successful check of a different plan.
+        self.import_limit_diagnostics = None
+
         # The cache key is only wanted when the cache is actually in play - a saving run always
         # simulates - so it is not computed otherwise. Building it as one tuple hash keeps the
         # per-window hashing in C rather than looping in Python, which matters because this runs on
@@ -578,6 +584,19 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
         predict_pv_power = {}
         predict_state = {}
         predict_grid_power = {}
+        import_diagnostics = None
+        if save and self.import_limit > 0:
+            import_diagnostics = {
+                "limit_w": self.import_limit,
+                "peak_w": 0.0,
+                "excess_kwh": 0.0,
+                "minutes_over": 0,
+                "first_exceeded": None,
+                "step_minutes": step,
+                "checked_minutes": 0,
+                "start": (self.midnight_utc + timedelta(minutes=self.minutes_now)).strftime(TIME_FORMAT),
+                "end": (self.midnight_utc + timedelta(minutes=self.minutes_now + self.forecast_minutes)).strftime(TIME_FORMAT),
+            }
         predict_load_power = {}
         predict_iboost = {}
         predict_carbon_g = {}
@@ -1361,6 +1380,21 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
                 # The car in a dispatch that has gone away is met by the grid (or PV surplus), never the battery
                 diff += car_gone_kwh
 
+            # Observe the raw whole-site AC balance, before presentation rounding. Do not clamp
+            # grid demand, invent battery discharge, or change a candidate's cost. This covers the
+            # full displayed forecast, including steps after the optimiser's end_record boundary.
+            if import_diagnostics is not None:
+                # A modelled car outside the house CT is absent from diff; it still shares the
+                # grid connection. Net it against any surplus, without counting metered cars twice.
+                import_w = max(diff + car_load_energy_bypass, 0.0) * MINUTE_WATT / step
+                import_diagnostics["peak_w"] = max(import_diagnostics["peak_w"], import_w)
+                import_diagnostics["checked_minutes"] += step
+                if import_w - self.import_limit > 1e-6:
+                    import_diagnostics["excess_kwh"] += (import_w - self.import_limit) * step / MINUTE_WATT
+                    import_diagnostics["minutes_over"] += step
+                    if import_diagnostics["first_exceeded"] is None:
+                        import_diagnostics["first_exceeded"] = stamp
+
             # Metric keep - pretend the battery is empty and you have to import instead of using the battery
             if best_soc_keep > 0 and soc <= best_soc_keep:
                 metric_keep += (best_soc_keep - soc) * import_rate * keep_minute_scaling * step / 60.0
@@ -1474,6 +1508,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
             minute += step
 
         hours_left = minute_left / 60.0
+        self.import_limit_diagnostics = import_diagnostics
         if self.debug_enable or save:
             self.hours_left = hours_left
             self.final_car_soc = final_car_soc

apps/predbat/prediction_kernel.py

@@ -897,6 +897,7 @@ def reset_kernel_run_state(pred):
     pred.predict_carbon_best = {}
     pred.predict_clipped_best = {}
     pred.predict_car_hold_best = {}
+    pred.import_limit_diagnostics = None
     pred.iboost_running = False
     pred.iboost_running_solar = False
     pred.iboost_running_full = False