Expose solar surplus power and force-export window state as sensors

apps/predbat/execute.py

@@ -1531,6 +1531,40 @@ def publish_inverter_data(self):
                 },
             )
 
+        # Power a load could take now without importing or draining the battery.
+        #
+        # The car's own draw is added back so the sensor keeps reading what is available for it
+        # rather than collapsing once the charger starts. Only when the charger is inside the CT
+        # clamp: outside it the grid reading never saw the car, so adding it would invent a surplus
+        # equal to the charger draw. prediction.py branches on the same flag.
+        car_add_back = self.car_charging_power if self.car_energy_reported_load else 0.0
+
+        # Subtracting battery discharge keeps battery power from being reported as solar. Battery
+        # charging is not added back, so a charging battery takes the surplus first.
+        solar_surplus = max(0.0, self.grid_power + car_add_back - max(0.0, self.battery_power))
+
+        # Nothing can be spare that was never generated. A grid sensor wired positive-on-import
+        # without grid_power_invert would otherwise make the surplus track the import, and this
+        # sensor switches real load on.
+        solar_surplus = min(solar_surplus, max(0.0, self.pv_power))
+        self.dashboard_item(
+            self.prefix + ".solar_surplus_power",
+            state=dp3(solar_surplus / 1000.0),
+            attributes={
+                "friendly_name": "Current solar surplus power",
+                "state_class": "measurement",
+                "unit_of_measurement": "kW",
+                "device_class": "power",
+                "icon": "mdi:solar-power",
+                "grid_power": dp3(self.grid_power / 1000.0),
+                "battery_power": dp3(self.battery_power / 1000.0),
+                "pv_power": dp3(self.pv_power / 1000.0),
+                "car_charging_power": dp3(self.car_charging_power / 1000.0),
+                "car_charging_power_configured": self.car_charging_power_configured,
+                "car_charging_power_included": bool(self.car_energy_reported_load),
+            },
+        )
+
     def publish_inverter_config(self):
         """
         Publish the static configuration the prediction runs from, aggregated over all the inverters

apps/predbat/output.py

@@ -3082,6 +3082,17 @@ def set_charge_export_status(self, isCharging, isExporting, isDemand):
         )
         self.dashboard_item("binary_sensor." + self.prefix + "_demand", state="on" if isDemand else "off", attributes={"friendly_name": "Predbat is in demand mode", "icon": "mdi:battery-arrow-up"})
 
+        # Taken from the plan, not from isExporting above. isExporting is off in read only mode,
+        # off during Hold exporting, and only on for the minutes an export is commanded, whereas an
+        # automation needs to know a slot is running for its whole duration including freeze export.
+        export_window_n = self.in_charge_window(self.export_window_best, self.minutes_now)
+        in_export_slot = self.set_export_window and export_window_n >= 0 and export_mode_of(self.export_limits_best[export_window_n]) != EXPORT_MODE_IDLE
+        self.dashboard_item(
+            "binary_sensor." + self.prefix + "_force_export_slot",
+            state="on" if in_export_slot else "off",
+            attributes={"friendly_name": "Predbat is in a force export slot", "icon": "mdi:transmission-tower-export"},
+        )
+
     def yesterday_reconstruct_car_slots(self, end_record, yesterday_load_step, minutes_now):
         """Rebuild car charging slots for yesterday and today-so-far, and subtract them from the load band.
 

apps/predbat/tests/test_solar_surplus.py

@@ -0,0 +1,218 @@
+# -----------------------------------------------------------------------------
+# Predbat Home Battery System
+# Copyright Trefor Southwell 2026 - All Rights Reserved
+# This application maybe used for personal use only and not for commercial use
+# -----------------------------------------------------------------------------
+# fmt off
+# pylint: disable=consider-using-f-string
+# pylint: disable=line-too-long
+# pylint: disable=attribute-defined-outside-init
+
+import copy
+
+from const import EXPORT_LIMIT_FREEZE, EXPORT_LIMIT_IDLE, EXPORT_MODE_TARGET, EXPORT_MODE_FREEZE, EXPORT_MODE_IDLE
+from utils import pack_export_limit
+from tests.test_infra import reset_inverter
+
+SURPLUS_ENTITY = ".solar_surplus_power"
+EXPORT_SLOT_ENTITY = "_force_export_slot"
+
+
+def check_surplus(my_predbat, name, grid_power, battery_power, car_charging_power, car_configured, expect_state, pv_power=20000, car_reported_load=True):
+    """Publish the inverter sensors for one power reading and check the surplus state
+
+    pv_power defaults high enough not to clip. Cases exercising the clamp pass their own.
+    """
+    my_predbat.grid_power = grid_power
+    my_predbat.battery_power = battery_power
+    my_predbat.pv_power = pv_power
+    my_predbat.car_charging_power = car_charging_power
+    my_predbat.car_charging_power_configured = car_configured
+    my_predbat.car_energy_reported_load = car_reported_load
+    my_predbat.publish_inverter_data()
+
+    entity_id = my_predbat.prefix + SURPLUS_ENTITY
+    item = my_predbat.ha_interface.dummy_items.get(entity_id)
+    if item is None:
+        print("ERROR: {} was not published for {}".format(entity_id, name))
+        return True
+    if item.get("state", None) != expect_state:
+        print("ERROR: {} state is {} expected {} for {}".format(entity_id, item.get("state", None), expect_state, name))
+        return True
+    return False
+
+
+def check_export_slot(my_predbat, name, minutes_now, export_window, export_limits, set_export_window, expect_on):
+    """Publish the charge/export status for one plan state and check the force export slot sensor"""
+    my_predbat.minutes_now = minutes_now
+    my_predbat.export_window_best = export_window
+    my_predbat.export_limits_best = export_limits
+    my_predbat.set_export_window = set_export_window
+    my_predbat.set_charge_export_status(False, False, True)
+
+    entity_id = "binary_sensor." + my_predbat.prefix + EXPORT_SLOT_ENTITY
+    item = my_predbat.ha_interface.dummy_items.get(entity_id)
+    if item is None:
+        print("ERROR: {} was not published for {}".format(entity_id, name))
+        return True
+    expect_state = "on" if expect_on else "off"
+    if item.get("state", None) != expect_state:
+        print("ERROR: {} state is {} expected {} for {}".format(entity_id, item.get("state", None), expect_state, name))
+        return True
+    return False
+
+
+def run_export_slot_tests(my_predbat):
+    """
+    binary_sensor.predbat_force_export_slot reports whether the plan has a force export window
+    running, so an automation can leave the solar for Predbat to sell.
+    """
+    failed = False
+    saved = {key: getattr(my_predbat, key) for key in ["minutes_now", "export_window_best", "export_limits_best", "set_export_window"]}
+    saved_items = copy.deepcopy(my_predbat.ha_interface.dummy_items)
+    window = [{"start": 600, "end": 660}]
+
+    try:
+        print("Test: inside a force export window the sensor is on")
+        failed |= check_export_slot(my_predbat, "inside window", 630, window, [pack_export_limit(EXPORT_MODE_TARGET, 10)], True, True)
+
+        print("Test: a freeze export window counts, as Predbat is selling the solar")
+        failed |= check_export_slot(my_predbat, "freeze export", 630, window, [pack_export_limit(EXPORT_MODE_FREEZE)], True, True)
+
+        print("Test: a window left at the idle limit is not an export slot")
+        failed |= check_export_slot(my_predbat, "idle limit", 630, window, [pack_export_limit(EXPORT_MODE_IDLE)], True, False)
+
+        print("Test: the bare numbers an older saved plan decodes into read the same way")
+        failed |= check_export_slot(my_predbat, "legacy target", 630, window, [10.0], True, True)
+        failed |= check_export_slot(my_predbat, "legacy freeze", 630, window, [EXPORT_LIMIT_FREEZE], True, True)
+        failed |= check_export_slot(my_predbat, "legacy idle", 630, window, [EXPORT_LIMIT_IDLE], True, False)
+
+        print("Test: before and after the window the sensor is off")
+        failed |= check_export_slot(my_predbat, "before window", 599, window, [10.0], True, False)
+        failed |= check_export_slot(my_predbat, "after window", 660, window, [10.0], True, False)
+
+        print("Test: with export control disabled the sensor is off even inside a window")
+        failed |= check_export_slot(my_predbat, "export control off", 630, window, [10.0], False, False)
+
+        print("Test: with no windows planned the sensor is off")
+        failed |= check_export_slot(my_predbat, "no windows", 630, [], [], True, False)
+
+        print("Test: the charging/exporting/demand sensors still report the commanded state")
+        my_predbat.set_charge_export_status(True, False, False)
+        for entity, expect_state in [("_charging", "on"), ("_exporting", "off"), ("_demand", "off")]:
+            entity_id = "binary_sensor." + my_predbat.prefix + entity
+            item = my_predbat.ha_interface.dummy_items.get(entity_id)
+            if item is None or item.get("state", None) != expect_state:
+                print("ERROR: {} state is {} expected {}".format(entity_id, item.get("state", None) if item else None, expect_state))
+                failed = True
+    finally:
+        for key, value in saved.items():
+            setattr(my_predbat, key, value)
+        my_predbat.ha_interface.dummy_items.clear()
+        my_predbat.ha_interface.dummy_items.update(saved_items)
+
+    return failed
+
+
+def run_solar_surplus_tests(my_predbat):
+    """
+    predbat.solar_surplus_power reports the power a load could take now without importing or
+    draining the battery, so an automation can run a car or immersion heater on spare solar.
+    """
+    reset_inverter(my_predbat)
+    failed = False
+
+    saved = {key: getattr(my_predbat, key) for key in ["grid_power", "battery_power", "pv_power", "car_charging_power", "car_charging_power_configured", "car_energy_reported_load"]}
+    saved_items = copy.deepcopy(my_predbat.ha_interface.dummy_items)
+    try:
+        print("Test: with no car charging the surplus is the grid export")
+        failed |= check_surplus(my_predbat, "plain export", grid_power=2000, battery_power=0, car_charging_power=0, car_configured=False, expect_state=2.0)
+
+        print("Test: importing from the grid reports no surplus rather than a negative one")
+        failed |= check_surplus(my_predbat, "importing", grid_power=-1500, battery_power=0, car_charging_power=0, car_configured=False, expect_state=0.0)
+
+        print("Test: a charging car is added back, so the sensor does not collapse once the car starts")
+        failed |= check_surplus(my_predbat, "car charging", grid_power=200, battery_power=0, car_charging_power=7000, car_configured=True, expect_state=7.2)
+
+        # Cloud cover mid-charge: the battery is covering part of the car and the grid has swung
+        # to import. Without the subtraction this reads 4.0kW and the car keeps charging off it.
+        print("Test: battery discharge is not offered to the car as solar surplus")
+        failed |= check_surplus(my_predbat, "battery covering the car", grid_power=-3000, battery_power=3000, car_charging_power=7000, car_configured=True, expect_state=1.0)
+
+        print("Test: battery charging takes the surplus, it is not added back")
+        failed |= check_surplus(my_predbat, "battery charging", grid_power=1000, battery_power=-2000, car_charging_power=0, car_configured=False, expect_state=1.0)
+
+        # Outside the CT clamp the grid reading never saw the car, so the add-back would invent a
+        # surplus equal to the charger draw and the automation would never turn the car off.
+        print("Test: a charger outside the CT clamp is not added back")
+        failed |= check_surplus(my_predbat, "outside CT clamp", grid_power=0, battery_power=0, car_charging_power=7000, car_configured=True, expect_state=0.0, car_reported_load=False)
+
+        print("Test: inside the CT clamp the same reading does add the car back")
+        failed |= check_surplus(my_predbat, "inside CT clamp", grid_power=0, battery_power=0, car_charging_power=7000, car_configured=True, expect_state=7.0, car_reported_load=True)
+
+        # A grid sensor wired positive-on-import without grid_power_invert reads +8kW while the
+        # house imports 8kW. Unclamped the surplus would track the import.
+        print("Test: the surplus can never exceed generation")
+        failed |= check_surplus(my_predbat, "inverted grid sensor at night", grid_power=8000, battery_power=0, car_charging_power=0, car_configured=False, expect_state=0.0, pv_power=0)
+        failed |= check_surplus(my_predbat, "clamped to PV", grid_power=8000, battery_power=0, car_charging_power=0, car_configured=False, expect_state=3.0, pv_power=3000)
+
+        print("Test: a genuine surplus is not clipped by the clamp")
+        failed |= check_surplus(my_predbat, "surplus below PV", grid_power=2500, battery_power=0, car_charging_power=0, car_configured=False, expect_state=2.5, pv_power=4000)
+
+        print("Test: the components and whether a charger is configured are published as attributes")
+        my_predbat.grid_power = 500
+        my_predbat.battery_power = -250
+        my_predbat.pv_power = 2750
+        my_predbat.car_charging_power = 1500
+        my_predbat.car_charging_power_configured = True
+        my_predbat.car_energy_reported_load = True
+        my_predbat.publish_inverter_data()
+        item = my_predbat.ha_interface.dummy_items.get(my_predbat.prefix + SURPLUS_ENTITY)
+        expect = {
+            "state": 2.0,
+            "friendly_name": "Current solar surplus power",
+            "state_class": "measurement",
+            "unit_of_measurement": "kW",
+            "device_class": "power",
+            "icon": "mdi:solar-power",
+            "grid_power": 0.5,
+            "battery_power": -0.25,
+            "pv_power": 2.75,
+            "car_charging_power": 1.5,
+            "car_charging_power_configured": True,
+            "car_charging_power_included": True,
+        }
+        for key, value in expect.items():
+            if item.get(key, None) != value:
+                print("ERROR: {} {} is {} expected {}".format(my_predbat.prefix + SURPLUS_ENTITY, key, item.get(key, None), value))
+                failed = True
+
+        print("Test: the sensor is published even when no car charger is configured")
+        my_predbat.car_charging_power = 0
+        my_predbat.car_charging_power_configured = False
+        my_predbat.ha_interface.dummy_items.pop(my_predbat.prefix + SURPLUS_ENTITY, None)
+        my_predbat.publish_inverter_data()
+        item = my_predbat.ha_interface.dummy_items.get(my_predbat.prefix + SURPLUS_ENTITY)
+        if item is None:
+            print("ERROR: {} was not published with no car charger configured".format(my_predbat.prefix + SURPLUS_ENTITY))
+            failed = True
+        elif item.get("car_charging_power_configured", None) is not False:
+            print("ERROR: {} car_charging_power_configured is {} expected False".format(my_predbat.prefix + SURPLUS_ENTITY, item.get("car_charging_power_configured", None)))
+            failed = True
+
+        print("Test: car_charging_power_included reports whether the add-back was applied")
+        my_predbat.car_energy_reported_load = False
+        my_predbat.publish_inverter_data()
+        item = my_predbat.ha_interface.dummy_items.get(my_predbat.prefix + SURPLUS_ENTITY)
+        if item.get("car_charging_power_included", None) is not False:
+            print("ERROR: {} car_charging_power_included is {} expected False".format(my_predbat.prefix + SURPLUS_ENTITY, item.get("car_charging_power_included", None)))
+            failed = True
+    finally:
+        for key, value in saved.items():
+            setattr(my_predbat, key, value)
+        my_predbat.ha_interface.dummy_items.clear()
+        my_predbat.ha_interface.dummy_items.update(saved_items)
+
+    failed |= run_export_slot_tests(my_predbat)
+
+    return failed

apps/predbat/unit_test.py

@@ -32,6 +32,7 @@
 from tests.test_predict_pv_power import run_predict_pv_power_tests
 from tests.test_dashboard_device_class import test_dashboard_device_class
 from tests.test_inverter_config_sensor import test_inverter_config_sensor
+from tests.test_solar_surplus import run_solar_surplus_tests
 from tests.test_kernel_parity import run_kernel_parity_tests, run_model_kernel_tests
 from tests.test_prediction_batch import run_prediction_batch_tests
 from tests.test_kernel_static_cache import run_kernel_static_cache_tests
@@ -443,6 +444,7 @@ def main():
         ("predict_pv_power", run_predict_pv_power_tests, "predict_pv_power plan-interval scaling tests", False),
         ("dashboard_device_class", test_dashboard_device_class, "Dashboard sensor device_class regression tests (#3352)", False),
         ("inverter_config_sensor", test_inverter_config_sensor, "Aggregated static prediction inputs published as sensor.<prefix>_inverter_config", False),
+        ("solar_surplus", run_solar_surplus_tests, "Solar surplus power and force export slot sensor tests", False),
         ("model_kernel", run_model_kernel_tests, "Model tests run with the C++ prediction kernel enabled", False),
         ("kernel_parity", run_kernel_parity_tests, "C++ prediction kernel vs Python engine parity tests", False),
         ("prediction_batch", run_prediction_batch_tests, "Batched prediction fan-out tests", False),