Add opportunistic solar (sun-following) car charging model

.cspell/custom-dictionary-workspace.txt

@@ -116,6 +116,7 @@ energydataservice
 energythroughput
 epod
 euids
+evcc
 evergen
 evse
 exog
@@ -207,6 +208,7 @@ linebreak
 linestyle
 loadml
 loadmlpower
+loadpoint
 loadspower
 localfolder
 locationless

apps/predbat/config.py

@@ -749,6 +749,19 @@
         "enable": "num_cars",
         "enable_condition": "num_cars > 0",
     },
+    {
+        "name": "car_charging_solar_min_soc",
+        "friendly_name": "Car charging solar min soc",
+        "type": "input_number",
+        "min": 0,
+        "max": 100,
+        "step": 5,
+        "unit": "%",
+        "icon": "mdi:percent",
+        "default": 0.0,
+        "enable": "num_cars",
+        "enable_condition": "num_cars > 0",
+    },
     {
         "name": "calculate_export_oncharge",
         "oldname": "calculate_discharge_oncharge",
@@ -2158,6 +2171,12 @@
     "car_charging_soc": {"type": "sensor", "sensor_type": "float", "entries": "num_cars"},
     "car_charging_limit": {"type": "sensor", "sensor_type": "float", "entries": "num_cars"},
     "car_charging_exclusive": {"type": "boolean_list", "entries": "num_cars"},
+    "car_charging_solar": {"type": "boolean_list", "entries": "num_cars"},
+    "car_charging_plugged": {"type": "sensor|sensor_list", "sensor_type": "string", "entries": "num_cars"},
+    "car_charging_solar_max_power": {"type": "sensor|sensor_list", "sensor_type": "float", "entries": "num_cars"},
+    "car_charging_solar_min_power": {"type": "sensor|sensor_list", "sensor_type": "float", "entries": "num_cars"},
+    "car_charging_solar_power_step": {"type": "sensor|sensor_list", "sensor_type": "float", "entries": "num_cars"},
+    "car_charging_solar_limit": {"type": "sensor|sensor_list", "sensor_type": "float", "entries": "num_cars"},
     "carbon_intensity": {"type": "sensor", "sensor_type": "string"},
     "carbon_postcode": {"type": "string", "empty": False},
     "carbon_automatic": {"type": "boolean"},

apps/predbat/fetch.py

@@ -1106,6 +1106,12 @@ def fetch_sensor_data_car_planning(self):
                     self.log("Car {} on Octopus Intelligent, active plan for charge".format(car_n))
                 else:
                     self.log("Car {} on Octopus Intelligent, no active plan".format(car_n))
+            elif self.car_charging_solar[car_n] and not self.car_charging_planned[car_n]:
+                # Opportunistic solar charging with no departure plan: do not plan any grid/low-rate slots.
+                # The PV diversion is modelled directly in the prediction (see prediction.py), so the home
+                # battery forecast still accounts for the energy the car takes, without scheduling grid charging.
+                self.car_charging_slots[car_n] = []
+                self.log("Car {} solar charging only (no active plan): no grid slots planned, PV diversion handled in the forecast".format(car_n))
             elif self.car_charging_planned[car_n] or self.car_charging_now[car_n]:
                 self.log(
                     "Car {} plan charging from {} to {}, with slots {} from SoC {}% to {}%, ready by {}".format(
@@ -1915,10 +1921,17 @@ def get_car_charging_planned(self):
         self.car_charging_rate = [7.4 for c in range(max(self.num_cars, 1))]
         self.car_charging_slots = [[] for c in range(self.num_cars)]
         self.car_charging_exclusive = [False for c in range(self.num_cars)]
+        self.car_charging_solar = [False for c in range(self.num_cars)]
+        self.car_charging_plugged = [False for c in range(self.num_cars)]
+        self.car_charging_solar_max_power = [self.car_charging_rate[c] for c in range(self.num_cars)]
+        self.car_charging_solar_min_power = [0.0 for c in range(self.num_cars)]
+        self.car_charging_solar_power_step = [0.0 for c in range(self.num_cars)]
+        self.car_charging_solar_limit = [self.car_charging_limit[c] for c in range(self.num_cars)]
 
         self.car_charging_planned_response = self.get_arg("car_charging_planned_response", ["yes", "on", "enable", "true"])
         self.car_charging_now_response = self.get_arg("car_charging_now_response", ["yes", "on", "enable", "true"])
         self.car_charging_from_battery = self.get_arg("car_charging_from_battery")
+        self.car_charging_solar_min_soc = self.get_arg("car_charging_solar_min_soc", 0.0)
 
         # Car charging planned sensor
         for car_n in range(self.num_cars):
@@ -1954,6 +1967,36 @@ def get_car_charging_planned(self):
             self.car_charging_limit[car_n] = dp3((float(self.get_arg("car_charging_limit", 100.0, index=car_n)) * self.car_charging_battery_size[car_n]) / 100.0)
             self.car_charging_exclusive[car_n] = self.get_arg("car_charging_exclusive", False, index=car_n)
 
+            # Opportunistic solar (sun-following) charging - models PV diverted to the car by an external charger (e.g. EVCC)
+            self.car_charging_solar[car_n] = self.get_arg("car_charging_solar", False, index=car_n)
+
+            # Maximum diversion power - defaults to the configured car charging rate, but is uncapped (3-phase chargers exceed the rate slider limit)
+            self.car_charging_solar_max_power[car_n] = float(self.get_arg("car_charging_solar_max_power", self.car_charging_rate[car_n], index=car_n))
+
+            # Minimum power needed before the charger will start diverting (e.g. 3-phase 6A)
+            self.car_charging_solar_min_power[car_n] = float(self.get_arg("car_charging_solar_min_power", 0.0, index=car_n))
+
+            # Discrete charge power step (kW). Real chargers only switch in whole current steps (e.g. 1A = ~0.69kW on 3-phase),
+            # so they leave a small surplus remainder. 0 = continuous (no quantisation).
+            self.car_charging_solar_power_step[car_n] = float(self.get_arg("car_charging_solar_power_step", 0.0, index=car_n))
+
+            # SoC limit (%) the opportunistic solar diversion fills the car to, independent of the grid plan target
+            # (car_charging_limit). Defaults to the grid plan target when not configured. Stored in kWh like car_charging_limit.
+            solar_limit_pct = self.get_arg("car_charging_solar_limit", None, index=car_n)
+            if solar_limit_pct is None:
+                self.car_charging_solar_limit[car_n] = self.car_charging_limit[car_n]
+            else:
+                self.car_charging_solar_limit[car_n] = dp3((float(solar_limit_pct) * self.car_charging_battery_size[car_n]) / 100.0)
+
+            # Plugged-in status over the forecast horizon - falls back to "charging now" if no dedicated sensor is configured
+            plugged = self.get_arg("car_charging_plugged", None, index=car_n)
+            if plugged is None:
+                self.car_charging_plugged[car_n] = self.car_charging_now[car_n]
+            elif isinstance(plugged, str):
+                self.car_charging_plugged[car_n] = plugged.lower() in self.car_charging_now_response
+            else:
+                self.car_charging_plugged[car_n] = bool(plugged)
+
         if self.num_cars > 0:
             self.log(
                 "Cars {} charging from battery {} planned {}, charging_now {} smart {}, max_price {}{}, plan_time {}, battery size {}kWh, limit {}%, rate {}kW, exclusive {}".format(

apps/predbat/output.py

@@ -1377,10 +1377,19 @@ def publish_html_plan(self, pv_forecast_minute_step, pv_forecast_minute_step10,
             # Car charging?
             if self.num_cars > 0:
                 car_charging_kwh = self.car_charge_slot_kwh(minute_start, minute_end)
-                car_total += car_charging_kwh
+                # Opportunistic solar diversion modelled in the forecast (cumulative kWh, like iBoost)
+                car_solar_amount = self.predict_car_solar_best.get(minute_relative_start, 0)
+                car_solar_amount_end = self.predict_car_solar_best.get(minute_relative_slot_end, car_solar_amount)
+                car_solar_change = max(car_solar_amount_end - car_solar_amount, 0.0)
+                car_total += car_charging_kwh + car_solar_change
                 if car_charging_kwh > 0.0:
-                    car_charging_str = str(car_charging_kwh)
-                    car_color = "FFFF00"
+                    # Planned (grid) charging - shown yellow, includes any solar diverted in the same slot
+                    car_charging_str = str(dp2(car_charging_kwh + car_solar_change))
+                    car_color = "#FFFF00"
+                elif car_solar_change > 0.0:
+                    # Pure opportunistic solar diversion - shown green
+                    car_charging_str = str(dp2(car_solar_change))
+                    car_color = "#AEF8A0"
                 else:
                     car_charging_str = "⚊"
                     car_color = "#FFFFFF"
@@ -1559,7 +1568,9 @@ def publish_html_plan(self, pv_forecast_minute_step, pv_forecast_minute_step10,
                 json_row["extra_load_total"] = raw_extra_forecast_total
                 json_row["extra_color"] = extra_color
             if self.num_cars > 0:
-                json_row["car_charging"] = car_charging_kwh
+                # Include the modelled opportunistic solar diversion so the JSON/web plan view matches the HTML cell
+                json_row["car_charging"] = dp2(car_charging_kwh + car_solar_change)
+                json_row["car_solar"] = dp2(car_solar_change)
                 json_row["car_color"] = car_color
             if self.iboost_enable:
                 json_row["iboost"] = iboost_amount

apps/predbat/plan.py

@@ -3521,6 +3521,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
                 self.predict_metric_best = pred.predict_metric_best
                 self.predict_carbon_best = pred.predict_carbon_best
                 self.predict_clipped_best = pred.predict_clipped_best
+                self.predict_car_solar_best = pred.predict_car_solar_best
 
             if save:
                 self.log(

apps/predbat/predbat.py

@@ -361,6 +361,7 @@ def reset(self):
         self.predict_soc = {}
         self.predict_soc_best = {}
         self.predict_iboost_best = {}
+        self.predict_car_solar_best = {}
         self.predict_metric_best = {}
         self.metric_min_improvement = 0.0
         self.metric_min_improvement_export = 0.1
@@ -457,6 +458,13 @@ def reset(self):
         self.car_charging_soc_next = [None]
         self.car_charging_rate = [7.4]
         self.car_charging_loss = 1.0
+        self.car_charging_solar = [False]
+        self.car_charging_plugged = [False]
+        self.car_charging_solar_max_power = [7.4]
+        self.car_charging_solar_min_power = [0.0]
+        self.car_charging_solar_power_step = [0.0]
+        self.car_charging_solar_limit = [100.0]
+        self.car_charging_solar_min_soc = 0.0
         self.export_window = []
         self.export_limits = []
         self.export_limits_best = []

apps/predbat/prediction.py

@@ -134,6 +134,13 @@ def __init__(self, base=None, pv_forecast_minute_step=None, pv_forecast_minute10
             self.car_charging_slots = base.car_charging_slots
             self.car_charging_limit = base.car_charging_limit
             self.car_charging_from_battery = base.car_charging_from_battery
+            self.car_charging_solar = base.car_charging_solar
+            self.car_charging_plugged = base.car_charging_plugged
+            self.car_charging_solar_max_power = base.car_charging_solar_max_power
+            self.car_charging_solar_min_power = base.car_charging_solar_min_power
+            self.car_charging_solar_power_step = base.car_charging_solar_power_step
+            self.car_charging_solar_limit = base.car_charging_solar_limit
+            self.car_charging_solar_min_soc = base.car_charging_solar_min_soc
             self.iboost_enable = base.iboost_enable
             self.iboost_on_export = base.iboost_on_export
             self.iboost_prevent_discharge = base.iboost_prevent_discharge
@@ -417,6 +424,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
         self.predict_iboost_best = {}
         self.predict_carbon_best = {}
         self.predict_clipped_best = {}
+        self.predict_car_solar_best = {}
         self.iboost_running = False
         self.iboost_running_solar = False
         self.iboost_running_full = False
@@ -488,6 +496,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
         iboost_running_solar = self.iboost_running_solar
         iboost_running_full = self.iboost_running_full
         car_load_energy_bypass = 0
+        car_solar_today = 0
 
         # Remove intersecting windows and optimise the data format of the charge/discharge window
         charge_limit, charge_window = remove_intersecting_windows(charge_limit, charge_window, export_limits, export_window)
@@ -647,6 +656,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
                     self.predict_iboost_best[minute] = round(iboost_today_kwh, 2)
                     self.predict_carbon_best[minute] = round(carbon_g, 0)
                     self.predict_clipped_best[minute] = round(clipped_today, 2)
+                    self.predict_car_solar_best[minute] = round(car_solar_today, 2)
             else:
                 stamp = ""
 
@@ -678,6 +688,44 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
 
             # Simulate car charging
             if car_enable:
+                # Opportunistic solar (sun-following) diversion model - applied BEFORE any planned grid charging so
+                # that free solar is used first and a planned grid top-up only covers the remainder (mirrors EVCC).
+                # The car takes the PV left after the house load is served (true surplus), once the home battery is
+                # above the configured priority SoC, capped at its own solar limit (independent of the grid plan
+                # target). Modelling only - Predbat does not control the car, it only reflects the diverted energy.
+                for car_n in range(self.num_cars):
+                    if self.car_charging_solar[car_n] and self.car_charging_plugged[car_n] and pv_now > 0 and car_soc[car_n] < self.car_charging_solar_limit[car_n]:
+                        # Home battery priority: only divert to the car once the home battery SoC is above the threshold
+                        if soc_max <= 0 or (soc * 100.0 / soc_max) >= self.car_charging_solar_min_soc:
+                            # Only the PV left after the house load is served is available to the car
+                            surplus = max(pv_now - load_yesterday, 0)
+                            # Available charge power (kW), capped at the maximum diversion power
+                            avail_power = min(surplus * 60.0 / step, self.car_charging_solar_max_power[car_n])
+                            min_power = self.car_charging_solar_min_power[car_n]
+                            power_step = self.car_charging_solar_power_step[car_n]
+                            if avail_power < min_power:
+                                # Below the charger's minimum start power - nothing is diverted
+                                charge_power = 0
+                            elif power_step > 0:
+                                # Real chargers only switch in whole current steps (e.g. 1A), so they charge at the
+                                # largest discrete level at or below the surplus, leaving a small remainder to the battery/export
+                                charge_power = min_power + int((avail_power - min_power) / power_step) * power_step
+                            else:
+                                charge_power = avail_power
+                            car_solar_amount = charge_power * step / 60.0
+                            if car_solar_amount > 0:
+                                # Cap by remaining capacity to the SOLAR limit (battery-side kWh -> PV-side draw via the charging loss)
+                                room = max(self.car_charging_solar_limit[car_n] - car_soc[car_n], 0)
+                                if self.car_charging_loss > 0:
+                                    car_solar_amount = min(car_solar_amount, room / self.car_charging_loss)
+                                else:
+                                    car_solar_amount = min(car_solar_amount, room)
+                                if car_solar_amount > 0:
+                                    pv_now -= car_solar_amount
+                                    car_soc[car_n] += car_solar_amount * self.car_charging_loss
+                                    car_solar_today += car_solar_amount
+
+                # Planned (grid) car charging - tops up toward the plan target (car_charging_limit), after solar
                 car_load, car_rate_slot = in_car_slot(minute_absolute, self.num_cars, self.car_charging_slots)
 
                 # Car charging?

apps/predbat/tests/test_infra.py

@@ -566,6 +566,13 @@ def simple_scenario(
     charge_car=0,
     car_charging_from_battery=True,
     car_energy_reported_load=True,
+    car_charging_solar=False,
+    car_solar_max_power=7.4,
+    car_solar_min_power=0.0,
+    car_solar_power_step=0.0,
+    car_solar_min_soc=0.0,
+    car_solar_limit=100.0,
+    assert_final_car_solar=None,
     iboost_solar=False,
     iboost_solar_excess=False,
     iboost_gas=False,
@@ -703,6 +710,15 @@ def simple_scenario(
     my_predbat.best_soc_keep_weight = keep_weight
     my_predbat.car_charging_soc[0] = car_soc
     my_predbat.car_charging_limit[0] = car_limit
+    my_predbat.car_charging_solar[0] = car_charging_solar
+    my_predbat.car_charging_plugged[0] = car_charging_solar
+    my_predbat.car_charging_solar_max_power[0] = car_solar_max_power
+    my_predbat.car_charging_solar_min_power[0] = car_solar_min_power
+    my_predbat.car_charging_solar_power_step[0] = car_solar_power_step
+    my_predbat.car_charging_solar_min_soc = car_solar_min_soc
+    if car_charging_solar:
+        # Solar diversion cap (separate from car_charging_limit, the grid plan target)
+        my_predbat.car_charging_solar_limit[0] = car_solar_limit
     my_predbat.inverter_can_charge_during_export = inverter_can_charge_during_export
     my_predbat.charge_scaling10 = charge_scaling10
 
@@ -734,9 +750,14 @@ def simple_scenario(
         pv10_step[minute] = pv_amount / (60 / 5) if pv10 else 0
         load10_step[minute] = load_amount / (60 / 5) if pv10 else 0
 
-    if charge_car:
+    if charge_car or car_charging_solar:
         my_predbat.num_cars = 1
-        my_predbat.car_charging_slots[0] = [{"start": my_predbat.minutes_now, "end": my_predbat.forecast_minutes + my_predbat.minutes_now, "kwh": charge_car * my_predbat.forecast_minutes / 60.0}]
+        if charge_car:
+            # Planned (grid) charging slot spanning the horizon; combine with car_charging_solar to test coexistence
+            my_predbat.car_charging_slots[0] = [{"start": my_predbat.minutes_now, "end": my_predbat.forecast_minutes + my_predbat.minutes_now, "kwh": charge_car * my_predbat.forecast_minutes / 60.0}]
+        else:
+            # Opportunistic solar car with no planned grid slots, diversion modelled in the prediction
+            my_predbat.car_charging_slots[0] = []
     else:
         my_predbat.num_cars = 0
         my_predbat.car_charging_slots[0] = []
@@ -838,6 +859,13 @@ def simple_scenario(
             print("ERROR: iBoost running full should be {}".format(assert_iboost_running_full))
         failed = True
 
+    if assert_final_car_solar is not None:
+        total_car_solar = prediction.predict_car_solar_best[max(prediction.predict_car_solar_best.keys())] if prediction.predict_car_solar_best else 0
+        if abs(total_car_solar - assert_final_car_solar) >= 0.1:
+            if not ignore_failed:
+                print("ERROR: Final car solar {} should be {}".format(total_car_solar, assert_final_car_solar))
+            failed = True
+
     if save != "none":
         total_clipped = prediction.predict_clipped_best[max(prediction.predict_clipped_best.keys())] if prediction.predict_clipped_best else 0
         if abs(total_clipped - assert_clipped) >= 0.9: